diff --git a/.gitignore b/.gitignore index ecc72d731..9bc2c1efc 100644 --- a/.gitignore +++ b/.gitignore @@ -6,7 +6,7 @@ /website/node_modules/ .DS_Store /.serverless/ -.phpunit.cache +.phpunit.result.cache node_modules/ package-lock.json /.claude diff --git a/src/Bref.php b/src/Bref.php index 38b705c52..f77d3d330 100644 --- a/src/Bref.php +++ b/src/Bref.php @@ -32,6 +32,11 @@ public static function events(): EventDispatcher return self::$eventDispatcher; } + public static function isRunningInStreamingMode(): bool + { + return (bool) getenv('BREF_STREAMED_MODE'); + } + /** * @internal Used by the Bref runtime */ diff --git a/src/Event/Http/HttpHandler.php b/src/Event/Http/HttpHandler.php index 6e61793c1..49734ac30 100644 --- a/src/Event/Http/HttpHandler.php +++ b/src/Event/Http/HttpHandler.php @@ -10,7 +10,7 @@ abstract class HttpHandler implements Handler abstract public function handleRequest(HttpRequestEvent $event, Context $context): HttpResponse; /** {@inheritDoc} */ - public function handle($event, Context $context): array + public function handle($event, Context $context): array|\Generator { // See https://bref.sh/docs/runtimes/http.html#cold-starts if (isset($event['warmer']) && $event['warmer'] === true) { diff --git a/src/Event/Http/HttpResponse.php b/src/Event/Http/HttpResponse.php index 2e0487453..6d1f27150 100644 --- a/src/Event/Http/HttpResponse.php +++ b/src/Event/Http/HttpResponse.php @@ -2,6 +2,9 @@ namespace Bref\Event\Http; +use Bref\Bref; +use Generator; + /** * Formats the response expected by AWS Lambda and the API Gateway integration. */ @@ -9,20 +12,21 @@ final class HttpResponse { private int $statusCode; private array $headers; - private string $body; + private string|\Generator $body; /** * @param array $headers */ - public function __construct(string $body, array $headers = [], int $statusCode = 200) + public function __construct(string|\Generator $body, array $headers = [], int $statusCode = 200) { $this->body = $body; $this->headers = $headers; $this->statusCode = $statusCode; } - public function toApiGatewayFormat(bool $multiHeaders = false, ?string $awsRequestId = null): array + public function toApiGatewayFormat(bool $multiHeaders = false, ?string $awsRequestId = null): array|\Generator { + $isStreamedMode = Bref::isRunningInStreamingMode(); $base64Encoding = (bool) getenv('BREF_BINARY_RESPONSES'); $headers = []; @@ -49,19 +53,28 @@ public function toApiGatewayFormat(bool $multiHeaders = false, ?string $awsReque // This is the format required by the AWS_PROXY lambda integration // See https://stackoverflow.com/questions/43708017/aws-lambda-api-gateway-error-malformed-lambda-proxy-response + + if ($isStreamedMode) { + return $this->yieldBody([ + 'statusCode' => $this->statusCode, + $headersKey => $headers, + ]); + } + return [ 'isBase64Encoded' => $base64Encoding, 'statusCode' => $this->statusCode, $headersKey => $headers, - 'body' => $base64Encoding ? base64_encode($this->body) : $this->body, + 'body' => $base64Encoding ? base64_encode($this->getBodyAsString()) : $this->getBodyAsString(), ]; } /** * See https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html#http-api-develop-integrations-lambda.response */ - public function toApiGatewayFormatV2(?string $awsRequestId = null): array + public function toApiGatewayFormatV2(?string $awsRequestId = null): array|\Generator { + $isStreamedMode = Bref::isRunningInStreamingMode(); $base64Encoding = (bool) getenv('BREF_BINARY_RESPONSES'); $headers = []; @@ -87,15 +100,61 @@ public function toApiGatewayFormatV2(?string $awsRequestId = null): array // serialized to `[]` (we want `{}`) so we force it to an empty object. $headers = empty($headers) ? new \stdClass : $headers; + if ($isStreamedMode) { + return $this->yieldBody([ + 'cookies' => $cookies, + 'statusCode' => $this->statusCode, + 'headers' => $headers, + ]); + } + return [ 'cookies' => $cookies, 'isBase64Encoded' => $base64Encoding, 'statusCode' => $this->statusCode, 'headers' => $headers, - 'body' => $base64Encoding ? base64_encode($this->body) : $this->body, + 'body' => $base64Encoding ? base64_encode($this->getBodyAsString()) : $this->getBodyAsString(), ]; } + /** + * Yields the metadata, the null-byte separator and the body chunks in the + * Lambda Streaming Format. + * + * @param array $metadata + */ + private function yieldBody(array $metadata): Generator + { + yield json_encode($metadata); + + yield "\0\0\0\0\0\0\0\0"; + + if ($this->body instanceof Generator) { + foreach ($this->body as $dataChunk) { + yield $dataChunk; + } + } else { + yield $this->body; + } + } + + private function getBodyAsString(): string + { + if ($this->body instanceof Generator) { + $dataChunk = ''; + + while ($this->body->valid()) { + $dataChunk .= $this->body->current(); + + $this->body->next(); + } + + return $dataChunk; + } + + return $this->body; + } + /** * See https://github.com/zendframework/zend-diactoros/blob/754a2ceb7ab753aafe6e3a70a1fb0370bde8995c/src/Response/SapiEmitterTrait.php#L96 */ diff --git a/src/FpmRuntime/FpmHandler.php b/src/FpmRuntime/FpmHandler.php index 6931b9caa..1248256f1 100644 --- a/src/FpmRuntime/FpmHandler.php +++ b/src/FpmRuntime/FpmHandler.php @@ -2,6 +2,7 @@ namespace Bref\FpmRuntime; +use Bref\Bref; use Bref\Context\Context; use Bref\Event\Http\HttpHandler; use Bref\Event\Http\HttpRequestEvent; @@ -10,6 +11,7 @@ use Bref\FpmRuntime\FastCgi\FastCgiRequest; use Bref\FpmRuntime\FastCgi\Timeout; use Exception; +use Generator; use hollodotme\FastCGI\Client; use hollodotme\FastCGI\Exceptions\TimedoutException; use hollodotme\FastCGI\Interfaces\ProvidesRequestData; @@ -129,23 +131,160 @@ public function __destruct() /** * Proxy the API Gateway event to PHP-FPM and return its response. * + * When the Lambda function is configured for streaming responses (`BREF_STREAMED_MODE=1`), + * the response body is streamed as PHP-FPM writes it. + * * @throws FastCgiCommunicationFailed * @throws Timeout * @throws Exception */ public function handleRequest(HttpRequestEvent $event, Context $context): HttpResponse { - $request = $this->eventToFastCgiRequest($event, $context); + $isHttpApiEvent = array_key_exists('http', $event->getRequestContext()); + $streamingEnabled = Bref::isRunningInStreamingMode() + && ! (bool) getenv('BREF_STREAM_NO_FIBER') + // Streaming responses are only supported for API Gateway HTTP APIs (not ALB nor REST APIs) + && $isHttpApiEvent; + + if ($streamingEnabled) { + return $this->handleStreamedRequest($event, $context); + } - // The script will timeout 1 second before the remaining time - // to allow some time for Bref/PHP-FPM to recover and cleanup - $margin = 1000; - $timeoutDelayInMs = max(1000, $context->getRemainingTimeInMillis() - $margin); + $response = $this->sendRequestToFastCgi($event, $context); + + if ($response === null) { + // Cannot happen: reading a FastCGI response always returns one + throw new RuntimeException('PHP-FPM returned no response'); + } + + $responseHeaders = $this->getResponseHeaders($response); + + // Extract the status code + if (isset($responseHeaders['status'])) { + $status = (int) (is_array($responseHeaders['status']) ? $responseHeaders['status'][0] : $responseHeaders['status']); + unset($responseHeaders['status']); + } + + $this->ensureStillRunning(); + + return new HttpResponse($response->getBody(), $responseHeaders, $status ?? 200); + } + + /** + * Stream the response of PHP-FPM back to Lambda as PHP-FPM writes it. + * + * The response body is a generator that yields the FastCGI STDOUT chunks as they + * are received. We use a Fiber to suspend the socket reading loop of the FastCGI + * client whenever a chunk arrives, and the generator resumes the Fiber each time + * Lambda asks for more data. + * + * @throws FastCgiCommunicationFailed + * @throws Timeout + * @throws Exception + */ + private function handleStreamedRequest(HttpRequestEvent $event, Context $context): HttpResponse + { + $responseFiber = new \Fiber(function () use ($event, $context): void { + $this->sendRequestToFastCgi( + $event, + $context, + function (string $stdOut = '', string $stdErr = ''): void { + if ($stdOut !== '') { + \Fiber::suspend(['stdout', $stdOut]); + } elseif ($stdErr !== '') { + \Fiber::suspend(['stderr', $stdErr]); + } + }, + false + ); + }); + + // Read chunks from the fiber until we have the full HTTP headers, so that we can + // return the status code and headers along with the streamed body + $outputAccumulator = ''; + $headerEnd = null; + while ($headerEnd === null && ! $responseFiber->isTerminated()) { + [$chunkType, $fiberChunk] = $this->nextFiberChunk($responseFiber); + + if ($fiberChunk === '') { + continue; + } + + if ($chunkType === 'stderr') { + fwrite(STDERR, $fiberChunk); + continue; + } + + $outputAccumulator .= $fiberChunk; + + $position = strpos($outputAccumulator, "\r\n\r\n"); + if ($position !== false) { + $headerEnd = $position; + } + } + + if ($headerEnd !== null) { + $headerBlock = substr($outputAccumulator, 0, $headerEnd); + $bodyStart = substr($outputAccumulator, $headerEnd + 4); + } else { + $headerBlock = $outputAccumulator; + $bodyStart = ''; + } + + [$status, $responseHeaders] = $this->parseResponseHeaders($headerBlock); + + $this->ensureStillRunning(); + + return new HttpResponse( + (function () use ($responseFiber, $bodyStart): Generator { + if ($bodyStart !== '') { + yield $bodyStart; + } + + while (! $responseFiber->isTerminated()) { + [$chunkType, $fiberChunk] = $this->nextFiberChunk($responseFiber); + + if ($chunkType === 'stderr') { + fwrite(STDERR, $fiberChunk); + } elseif ($fiberChunk !== '') { + // We must never yield an empty string: it would be interpreted + // as the end of the streamed response body + yield $fiberChunk; + } + } + })(), + $responseHeaders, + $status + ); + } + + /** + * Send the FastCGI request to PHP-FPM and read its response. + * + * When a $passThroughCallback is provided and $readResponse is false, the response is + * not accumulated: the callback is invoked with each chunk of data as it is received. + * + * @throws FastCgiCommunicationFailed + * @throws Timeout + */ + private function sendRequestToFastCgi( + HttpRequestEvent $event, + Context $context, + ?callable $passThroughCallback = null, + bool $readResponse = true, + ): ?ProvidesResponseData { + $request = $this->eventToFastCgiRequest($event, $context, $passThroughCallback); try { $socketId = $this->client->sendAsyncRequest($this->connection, $request); - $response = $this->client->readResponse($socketId, $timeoutDelayInMs); + if ($readResponse) { + return $this->client->readResponse($socketId, $this->getRequestTimeoutInMs($context)); + } + + $this->client->waitForResponse($socketId, $this->getRequestTimeoutInMs($context)); + + return null; } catch (TimedoutException) { $invocationId = $context->getAwsRequestId(); echo "$invocationId The PHP script timed out. Bref will now restart PHP-FPM to start from a clean slate and flush the PHP logs.\nTimeouts can happen for example when trying to connect to a remote API or database, if this happens continuously check for those.\nIf you are using a RDS database, read this: https://bref.sh/docs/environment/database.html#accessing-the-internet\n"; @@ -170,7 +309,7 @@ public function handleRequest(HttpRequestEvent $event, Context $context): HttpRe // - this is reported as a Lambda execution error ("error rate" metrics are accurate) // - the CloudWatch logs correctly reflect that an execution error occurred // - the 500 response is the same as if an exception happened in Bref - throw new Timeout($timeoutDelayInMs, $context->getAwsRequestId()); + throw new Timeout($this->getRequestTimeoutInMs($context), $context->getAwsRequestId()); } catch (Throwable $e) { printf( "Error communicating with PHP-FPM to read the HTTP response. Bref will restart PHP-FPM now. Original exception message: %s %s\n", @@ -184,18 +323,64 @@ public function handleRequest(HttpRequestEvent $event, Context $context): HttpRe throw new FastCgiCommunicationFailed; } + } - $responseHeaders = $this->getResponseHeaders($response); + private function getRequestTimeoutInMs(Context $context): int + { + // The script will timeout 1 second before the remaining time + // to allow some time for Bref/PHP-FPM to recover and cleanup + $margin = 1000; - // Extract the status code - if (isset($responseHeaders['status'])) { - $status = (int) (is_array($responseHeaders['status']) ? $responseHeaders['status'][0] : $responseHeaders['status']); - unset($responseHeaders['status']); + return max(1000, $context->getRemainingTimeInMillis() - $margin); + } + + /** + * Start or resume the fiber and return the next chunk it suspends with. + * + * @return array{0: string, 1: string} + */ + private function nextFiberChunk(\Fiber $fiber): array + { + if ($fiber->isTerminated()) { + return ['', '']; } - $this->ensureStillRunning(); + $result = $fiber->isStarted() || $fiber->isSuspended() ? $fiber->resume() : $fiber->start(); - return new HttpResponse($response->getBody(), $responseHeaders, $status ?? 200); + return is_array($result) ? $result : ['', '']; + } + + /** + * Parse a raw FastCGI header block into a status code and a list of headers. + * + * @return array{0: int, 1: array} + */ + private function parseResponseHeaders(string $headerBlock): array + { + $status = 200; + $headers = []; + + $normalized = str_replace(["\r\n", "\r"], "\n", $headerBlock); + foreach (explode("\n", $normalized) as $line) { + if (preg_match('/^Status:\s*(\d{3})/i', $line, $matches) === 1) { + $status = (int) $matches[1]; + continue; + } + + $separator = strpos($line, ':'); + if ($separator === false) { + continue; + } + + $name = strtolower(trim(substr($line, 0, $separator))); + $value = trim(substr($line, $separator + 1)); + + if ($name !== '') { + $headers[$name] = $value; + } + } + + return [$status, $headers]; } /** @@ -240,7 +425,7 @@ private function isReady(): bool return file_exists(self::SOCKET); } - private function eventToFastCgiRequest(HttpRequestEvent $event, Context $context): ProvidesRequestData + private function eventToFastCgiRequest(HttpRequestEvent $event, Context $context, ?callable $passThroughCallback = null): ProvidesRequestData { $request = new FastCgiRequest($event->getMethod(), $this->handler, $event->getBody()); $request->setRequestUri($event->getUri()); @@ -255,6 +440,10 @@ private function eventToFastCgiRequest(HttpRequestEvent $event, Context $context $request->setCustomVar('LAMBDA_INVOCATION_CONTEXT', json_encode($context, JSON_THROW_ON_ERROR)); $request->setCustomVar('LAMBDA_REQUEST_CONTEXT', json_encode($event->getRequestContext(), JSON_THROW_ON_ERROR)); + if ($passThroughCallback !== null) { + $request->addPassThroughCallbacks($passThroughCallback); + } + $contentType = $event->getContentType(); if ($contentType) { $request->setContentType($contentType); diff --git a/src/Runtime/LambdaRuntime.php b/src/Runtime/LambdaRuntime.php index 4243cac41..660963c76 100755 --- a/src/Runtime/LambdaRuntime.php +++ b/src/Runtime/LambdaRuntime.php @@ -8,6 +8,7 @@ use Bref\Event\Handler; use CurlHandle; use Exception; +use Generator; use JsonException; use Psr\Http\Server\RequestHandlerInterface; use RuntimeException; @@ -37,6 +38,7 @@ final class LambdaRuntime { private ?CurlHandle $curlHandleNext = null; private ?CurlHandle $curlHandleResult = null; + private ?CurlHandle $curlStreamedHandleResult = null; private string $apiUrl; private Invoker $invoker; private string $layer; @@ -97,7 +99,14 @@ public function processNextEvent(Handler | RequestHandlerInterface | callable $h $this->sendResponse($context->getAwsRequestId(), $result); } catch (Throwable $e) { - $this->signalFailure($context->getAwsRequestId(), $e); + if (isset($result) && $result instanceof Generator) { + // We cannot signal a failure once a streamed response has started: Lambda + // would reject the error report with an "InvalidStateTransition" error. + // The error is only logged, the caller will see a truncated response. + $this->logError($e, $context->getAwsRequestId()); + } else { + $this->signalFailure($context->getAwsRequestId(), $e); + } try { Bref::events()->afterInvoke($handler, $event, $context, null, $e); @@ -201,7 +210,12 @@ private function waitNextInvocation(): array private function sendResponse(string $invocationId, mixed $responseData): void { $url = "http://$this->apiUrl/2018-06-01/runtime/invocation/$invocationId/response"; - $this->postJson($url, $responseData); + + if ($responseData instanceof Generator) { + $this->postStreamed($url, $responseData); + } else { + $this->postJson($url, $responseData); + } } /** @@ -281,6 +295,93 @@ public function failInitialization( exit(1); } + /** + * @param string[] $headers + * @throws Exception + * @throws ResponseTooBig + */ + private function postStreamed(string $url, Generator $data, array $headers = []): void + { + if ($this->curlStreamedHandleResult === null) { + $this->curlStreamedHandleResult = curl_init(); + curl_setopt($this->curlStreamedHandleResult, CURLOPT_UPLOAD, true); + curl_setopt($this->curlStreamedHandleResult, CURLOPT_CUSTOMREQUEST, 'POST'); + curl_setopt($this->curlStreamedHandleResult, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); + curl_setopt($this->curlStreamedHandleResult, CURLOPT_RETURNTRANSFER, true); + curl_setopt($this->curlStreamedHandleResult, CURLOPT_INFILESIZE, -1); + } + + curl_setopt($this->curlStreamedHandleResult, CURLOPT_URL, $url); + curl_setopt($this->curlStreamedHandleResult, CURLOPT_HTTPHEADER, [ + 'Lambda-Runtime-Function-Response-Mode: streaming', + 'Content-Type: application/vnd.awslambda.http-integration-response', + 'Transfer-Encoding: chunked', + ...$headers, + ]); + + $buffer = ''; + /* + * We use Fibers so we can suspend the yields and read data as needed. + * That way we don't block the response as more data comes. + */ + $fiber = new \Fiber( + function () use (&$data): void { + foreach ($data as $dataChunk) { + \Fiber::suspend((string) $dataChunk); + } + + \Fiber::suspend(PHP_INT_MIN); + } + ); + + curl_setopt( + $this->curlStreamedHandleResult, + CURLOPT_READFUNCTION, + function ($ch, $fd, $length) use (&$fiber, &$buffer) { + if ($buffer === '') { + if ($fiber->isStarted() || $fiber->isSuspended()) { + $fiberChunk = $fiber->resume(); + } elseif (! $fiber->isTerminated()) { + $fiberChunk = $fiber->start(); + } else { + $fiberChunk = PHP_INT_MIN; + } + + if ($fiberChunk !== PHP_INT_MIN) { + $buffer .= $fiberChunk; + } + } + + $chunk = substr($buffer, 0, $length); + $buffer = substr($buffer, strlen($chunk)); + + return $chunk; + } + ); + + $body = curl_exec($this->curlStreamedHandleResult); + + $statusCode = curl_getinfo($this->curlStreamedHandleResult, CURLINFO_HTTP_CODE); + if ($statusCode >= 400) { + // Re-open the connection in case of failure to start from a clean state + $this->closeCurlStreamedHandleResult(); + + if ($statusCode === 413) { + throw new ResponseTooBig; + } + + try { + $error = json_decode($body, true, 512, JSON_THROW_ON_ERROR); + $errorMessage = "{$error['errorType']}: {$error['errorMessage']}"; + } catch (JsonException) { + // In case we didn't get any JSON + $errorMessage = 'unknown error'; + } + + throw new Exception("Error $statusCode while calling the Lambda runtime API: $errorMessage"); + } + } + /** * @param string[] $headers * @throws Exception @@ -354,6 +455,13 @@ private function closeCurlHandleResult(): void } } + private function closeCurlStreamedHandleResult(): void + { + if ($this->curlStreamedHandleResult !== null) { + $this->curlStreamedHandleResult = null; + } + } + /** * Ping a Bref server with a statsd request. * diff --git a/tests/Event/Http/HttpHandlerTest.php b/tests/Event/Http/HttpHandlerTest.php new file mode 100644 index 000000000..20c686743 --- /dev/null +++ b/tests/Event/Http/HttpHandlerTest.php @@ -0,0 +1,101 @@ +createHandler('

Hello world!

'); + + $result = $handler->handle( + json_decode(file_get_contents(__DIR__ . '/Fixture/ag-v1-simple.json'), true, 512, JSON_THROW_ON_ERROR), + Context::fake() + ); + + self::assertInstanceOf(Generator::class, $result); + + self::assertSame([ + 'statusCode' => 200, + 'headers' => [], + ], json_decode($result->current(), true, 512, JSON_THROW_ON_ERROR)); + + $result->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $result->current()); + + $result->next(); + self::assertSame('

Hello world!

', $result->current()); + } + + public function test streamed response in API Gateway v2() + { + $handler = $this->createHandler('

Hello world!

'); + + $result = $handler->handle( + json_decode(file_get_contents(__DIR__ . '/Fixture/ag-v2-simple.json'), true, 512, JSON_THROW_ON_ERROR), + Context::fake() + ); + + self::assertInstanceOf(Generator::class, $result); + + self::assertSame([ + 'cookies' => [], + 'statusCode' => 200, + 'headers' => [], + ], json_decode($result->current(), true, 512, JSON_THROW_ON_ERROR)); + + $result->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $result->current()); + + $result->next(); + self::assertSame('

Hello world!

', $result->current()); + } + + public function test warmer invocations are handled() + { + $handler = $this->createHandler(''); + + $result = $handler->handle(['warmer' => true], Context::fake()); + + self::assertSame(['Lambda is warm'], $result); + } + + private function createHandler(string $body): HttpHandler + { + return new class($body) extends HttpHandler { + private string $body; + + public function __construct(string $body) + { + $this->body = $body; + } + + public function handleRequest(HttpRequestEvent $event, Context $context): HttpResponse + { + return new HttpResponse($this->body); + } + }; + } +} diff --git a/tests/Event/Http/HttpResponseStreamedTest.php b/tests/Event/Http/HttpResponseStreamedTest.php new file mode 100644 index 000000000..3897261fd --- /dev/null +++ b/tests/Event/Http/HttpResponseStreamedTest.php @@ -0,0 +1,190 @@ +Hello world!

', [ + 'Content-Type' => 'text/html; charset=utf-8', + ]); + + $generator = $response->toApiGatewayFormat(); + self::assertInstanceOf(Generator::class, $generator); + + self::assertSame([ + 'statusCode' => 200, + 'headers' => [ + 'Content-Type' => 'text/html; charset=utf-8', + ], + ], json_decode($generator->current(), true, 512, JSON_THROW_ON_ERROR)); + + $generator->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $generator->current()); + + $generator->next(); + self::assertSame('

Hello world!

', $generator->current()); + } + + public function test streamed response in API Gateway v2 format() + { + $response = new HttpResponse('

Hello world!

', [ + 'Content-Type' => 'text/html; charset=utf-8', + ]); + + $generator = $response->toApiGatewayFormatV2(); + self::assertInstanceOf(Generator::class, $generator); + + self::assertSame([ + 'cookies' => [], + 'statusCode' => 200, + 'headers' => [ + 'Content-Type' => 'text/html; charset=utf-8', + ], + ], json_decode($generator->current(), true, 512, JSON_THROW_ON_ERROR)); + + $generator->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $generator->current()); + + $generator->next(); + self::assertSame('

Hello world!

', $generator->current()); + } + + public function test streamed response with multi value headers() + { + $response = new HttpResponse('', [ + 'foo' => ['bar', 'baz'], + ]); + + $generator = $response->toApiGatewayFormat(true); + self::assertInstanceOf(Generator::class, $generator); + + self::assertSame([ + 'statusCode' => 200, + 'multiValueHeaders' => [ + 'Foo' => ['bar', 'baz'], + ], + ], json_decode($generator->current(), true, 512, JSON_THROW_ON_ERROR)); + + $generator->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $generator->current()); + + $generator->next(); + self::assertSame('', $generator->current()); + } + + public function test streamed response with cookies() + { + $response = new HttpResponse('', [ + 'set-cookie' => ['foo', 'bar'], + ]); + + $generator = $response->toApiGatewayFormatV2(); + self::assertInstanceOf(Generator::class, $generator); + + self::assertSame([ + 'cookies' => ['foo', 'bar'], + 'statusCode' => 200, + 'headers' => [], + ], json_decode($generator->current(), true, 512, JSON_THROW_ON_ERROR)); + } + + public static function provideStreamedBody(): iterable + { + yield 'single chunk' => [['

Hello world!

']]; + yield 'multiple chunks' => [['Hello', ' ', 'world!']]; + } + + /** + * @param array $chunks + * + * @dataProvider provideStreamedBody + */ + public function test streamed response with a generator body in API Gateway v1 format(array $chunks) + { + $response = new HttpResponse($this->createBodyGenerator($chunks), [ + 'Content-Type' => 'text/html; charset=utf-8', + ], 201); + + $generator = $response->toApiGatewayFormat(); + self::assertInstanceOf(Generator::class, $generator); + + self::assertSame([ + 'statusCode' => 201, + 'headers' => [ + 'Content-Type' => 'text/html; charset=utf-8', + ], + ], json_decode($generator->current(), true, 512, JSON_THROW_ON_ERROR)); + + $generator->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $generator->current()); + + foreach ($chunks as $chunk) { + $generator->next(); + self::assertSame($chunk, $generator->current()); + } + } + + /** + * @param array $chunks + * + * @dataProvider provideStreamedBody + */ + public function test streamed response with a generator body in API Gateway v2 format(array $chunks) + { + $response = new HttpResponse($this->createBodyGenerator($chunks), [ + 'Content-Type' => 'text/html; charset=utf-8', + ], 201); + + $generator = $response->toApiGatewayFormatV2(); + self::assertInstanceOf(Generator::class, $generator); + + self::assertSame([ + 'cookies' => [], + 'statusCode' => 201, + 'headers' => [ + 'Content-Type' => 'text/html; charset=utf-8', + ], + ], json_decode($generator->current(), true, 512, JSON_THROW_ON_ERROR)); + + $generator->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $generator->current()); + + foreach ($chunks as $chunk) { + $generator->next(); + self::assertSame($chunk, $generator->current()); + } + } + + /** + * @param array $chunks + */ + private function createBodyGenerator(array $chunks): Generator + { + foreach ($chunks as $chunk) { + yield $chunk; + } + } +} diff --git a/tests/Event/Http/HttpResponseTest.php b/tests/Event/Http/HttpResponseTest.php index 9eb36c7bf..186a40890 100644 --- a/tests/Event/Http/HttpResponseTest.php +++ b/tests/Event/Http/HttpResponseTest.php @@ -192,4 +192,42 @@ public function test response with multiple cookies() 'body' => '', ], $response->toApiGatewayFormatV2()); } + + public function test non streamed response with a generator body concatenates the chunks() + { + putenv('BREF_STREAMED_MODE=0'); + + $response = new HttpResponse($this->createBodyGenerator(), [ + 'Content-Type' => 'text/html; charset=utf-8', + ]); + + self::assertEquals([ + 'isBase64Encoded' => false, + 'statusCode' => 200, + 'headers' => [ + 'Content-Type' => 'text/html; charset=utf-8', + ], + 'body' => '

Hello world!

', + ], $response->toApiGatewayFormat()); + + $responseV2 = new HttpResponse($this->createBodyGenerator(), [ + 'Content-Type' => 'text/html; charset=utf-8', + ]); + + self::assertEquals([ + 'cookies' => [], + 'isBase64Encoded' => false, + 'statusCode' => 200, + 'headers' => [ + 'Content-Type' => 'text/html; charset=utf-8', + ], + 'body' => '

Hello world!

', + ], $responseV2->toApiGatewayFormatV2()); + } + + private function createBodyGenerator(): \Generator + { + yield '

Hello '; + yield 'world!

'; + } } diff --git a/tests/FpmRuntime/FpmHandlerStreamedTest.php b/tests/FpmRuntime/FpmHandlerStreamedTest.php new file mode 100644 index 000000000..e1804769e --- /dev/null +++ b/tests/FpmRuntime/FpmHandlerStreamedTest.php @@ -0,0 +1,171 @@ +fakeContext = new Context('abc', time(), 'abc', 'abc'); + } + + public function tearDown(): void + { + $this->fpm?->stop(); + putenv('BREF_STREAMED_MODE=0'); + } + + private function startFpm(string $fixture = 'streaming.php'): void + { + $this->fpm = new FpmHandler(__DIR__ . "/fixtures/$fixture", __DIR__ . '/fixtures/php-fpm.conf'); + $this->fpm->start(); + } + + /** + * Returns an API Gateway HTTP API (version 2.0) event. + */ + private function getHttpApiEvent(): array + { + return [ + 'version' => '2.0', + 'routeKey' => 'ANY /{proxy+}', + 'rawPath' => '/', + 'rawQueryString' => '', + 'headers' => [], + 'requestContext' => [ + 'accountId' => '123456789012', + 'apiId' => 'api-id', + 'domainName' => 'id.execute-api.us-east-1.amazonaws.com', + 'http' => [ + 'method' => 'GET', + 'path' => '/', + 'protocol' => 'HTTP/1.1', + 'sourceIp' => '127.0.0.1', + 'userAgent' => 'Test', + ], + 'requestId' => 'id', + 'routeKey' => 'ANY /{proxy+}', + 'stage' => '$default', + 'time' => '12/Mar/2020:19:03:58 +0000', + 'timeEpoch' => 1583348638391, + ], + ]; + } + + public function test streamed response is returned as a generator() + { + $this->startFpm(); + + $result = $this->fpm->handle($this->getHttpApiEvent(), $this->fakeContext); + + self::assertInstanceOf(Generator::class, $result); + } + + public function test streamed response sends chunks incrementally() + { + $this->startFpm(); + + $result = $this->fpm->handle($this->getHttpApiEvent(), $this->fakeContext); + self::assertInstanceOf(Generator::class, $result); + + // The first yield of the streamed format is the response metadata (status code and headers) + $result->rewind(); + $metadata = json_decode((string) $result->current(), true, 512, JSON_THROW_ON_ERROR); + $result->next(); + + self::assertSame(201, $metadata['statusCode']); + self::assertSame('streamed-value', $metadata['headers']['X-Custom-Header']); + + // The body is streamed chunk by chunk: the first chunk arrives as soon as + // PHP-FPM starts emitting it, while the total response takes longer than + // one blocking FastCGI read + $start = microtime(true); + $firstChunkTime = null; + $body = ''; + while ($result->valid()) { + $body .= (string) $result->current(); + $firstChunkTime ??= microtime(true); + $result->next(); + } + $totalTime = microtime(true) - $start; + + // The streamed body is prefixed with the null-byte separator of the Lambda Streaming Format + self::assertSame("\0\0\0\0\0\0\0\0" . 'chunk-1chunk-2chunk-3', $body); + self::assertLessThan(0.15, $firstChunkTime - $start); + self::assertGreaterThan(0.25, $totalTime); + } + + public function test streamed response keeps the response headers() + { + $this->startFpm(); + + $result = $this->fpm->handle($this->getHttpApiEvent(), $this->fakeContext); + self::assertInstanceOf(Generator::class, $result); + + $result->rewind(); + $metadata = json_decode((string) $result->current(), true, 512, JSON_THROW_ON_ERROR); + + self::assertSame('text/plain;charset=UTF-8', $metadata['headers']['Content-Type']); + } + + public function test response is not streamed when streamed mode is disabled() + { + putenv('BREF_STREAMED_MODE=0'); + + $this->startFpm(); + + $result = $this->fpm->handle($this->getHttpApiEvent(), $this->fakeContext); + + self::assertIsArray($result); + self::assertSame(201, $result['statusCode']); + self::assertSame('chunk-1chunk-2chunk-3', $result['body']); + self::assertSame('streamed-value', $result['headers']['X-Custom-Header']); + } + + public function test response is not streamed from FPM for non HTTP API events even in streamed mode() + { + $this->startFpm(); + + // ALB events do not contain `requestContext.http`: the response is not streamed + // from PHP-FPM, but the streamed format is still applied by the runtime + $albEvent = [ + 'requestContext' => [ + 'elb' => ['targetGroupArn' => 'arn:aws:elasticloadbalancing:...'], + ], + 'httpMethod' => 'GET', + 'path' => '/', + 'headers' => [], + 'body' => '', + 'isBase64Encoded' => false, + ]; + + $result = $this->fpm->handle($albEvent, $this->fakeContext); + + self::assertInstanceOf(Generator::class, $result); + + $result->rewind(); + $metadata = json_decode((string) $result->current(), true, 512, JSON_THROW_ON_ERROR); + self::assertSame(201, $metadata['statusCode']); + + $result->next(); + self::assertSame("\0\0\0\0\0\0\0\0", $result->current()); + + $result->next(); + self::assertSame('chunk-1chunk-2chunk-3', $result->current()); + } +} diff --git a/tests/FpmRuntime/fixtures/streaming.php b/tests/FpmRuntime/fixtures/streaming.php new file mode 100644 index 000000000..2b784017f --- /dev/null +++ b/tests/FpmRuntime/fixtures/streaming.php @@ -0,0 +1,21 @@ +assertSame('1', $requestId); $this->assertSame('Root=1-67891233-abcdef012345678912345678', $traceId); } + + public function test streamed response is sent with the streaming response mode() + { + putenv('BREF_STREAMED_MODE=1'); + try { + $this->givenAnEvent(['Hello' => 'world!']); + + $this->runtime->processNextEvent(function () { + $generator = (function () { + yield '

Hello world!

'; + })(); + + return (new HttpResponse($generator, [ + 'Content-Type' => 'text/html; charset=utf-8', + ]))->toApiGatewayFormatV2(); + }); + } finally { + putenv('BREF_STREAMED_MODE=0'); + } + + $requests = Server::received(); + $this->assertCount(2, $requests); + [$eventRequest, $eventStreamResponse] = $requests; + + $this->assertSame('GET', $eventRequest->getMethod()); + $this->assertSame('http://localhost:8126/2018-06-01/runtime/invocation/next', $eventRequest->getUri()->__toString()); + + $this->assertSame('POST', $eventStreamResponse->getMethod()); + $this->assertSame('http://localhost:8126/2018-06-01/runtime/invocation/1/response', $eventStreamResponse->getUri()->__toString()); + + $this->assertSame('streaming', $eventStreamResponse->getHeaderLine('lambda-runtime-function-response-mode')); + $this->assertSame('chunked', $eventStreamResponse->getHeaderLine('transfer-encoding')); + $this->assertSame('application/vnd.awslambda.http-integration-response', $eventStreamResponse->getHeaderLine('content-type')); + + $this->assertStringContainsString('"statusCode":200', (string) $eventStreamResponse->getBody()); + $this->assertStringContainsString("\0\0\0\0\0\0\0\0", (string) $eventStreamResponse->getBody()); + $this->assertStringContainsString('

Hello world!

', (string) $eventStreamResponse->getBody()); + } + + public function test response with a string body is streamed in streamed mode() + { + putenv('BREF_STREAMED_MODE=1'); + try { + $this->givenAnEvent(['Hello' => 'world!']); + + $this->runtime->processNextEvent(function () { + return (new HttpResponse('

Hello world!

', [ + 'Content-Type' => 'text/html; charset=utf-8', + ]))->toApiGatewayFormatV2(); + }); + } finally { + putenv('BREF_STREAMED_MODE=0'); + } + + $requests = Server::received(); + $this->assertCount(2, $requests); + [$eventRequest, $eventResponse] = $requests; + + $this->assertSame('GET', $eventRequest->getMethod()); + + $this->assertSame('POST', $eventResponse->getMethod()); + $this->assertSame('streaming', $eventResponse->getHeaderLine('lambda-runtime-function-response-mode')); + $this->assertStringContainsString("\0\0\0\0\0\0\0\0", (string) $eventResponse->getBody()); + $this->assertStringContainsString('

Hello world!

', (string) $eventResponse->getBody()); + } + + public function test an error during streaming does not signal a failure() + { + $this->givenAnEvent(['Hello' => 'world!']); + + $this->runtime->processNextEvent(function () { + return (function () { + yield 'chunk-1'; + throw new Exception('Boom during streaming'); + })(); + }); + + // No error response should be sent once the streaming has started + foreach (Server::received() as $request) { + $this->assertNotSame( + 'http://localhost:8126/2018-06-01/runtime/invocation/1/error', + $request->getUri()->__toString(), + 'an error must not be signaled once streaming has started' + ); + } + + $this->assertErrorInLogs('Exception', 'Boom during streaming'); + } }