diff --git a/README.md b/README.md index 551ff08..7136832 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,24 @@ A `POST /users` with `{"name": 7}` never reaches the handler; the router answers {"error": "Body does not describe User.", "errors": {"name": "expected string, got int", "email": "missing"}} ``` +## Calling another service + +`send` is the client: a `Request` in, an `IO` out, its body a stream read from the connection as it is compiled. A newline-delimited JSON export decodes straight into entities, one line at a time: + +```php +use function Phunkie\Http4p\Functions\client\send; +use function Phunkie\Http4p\Functions\decodeLines; + +send(Request(Method::GET, 'http://catalogue.internal/books/export')) + ->flatMap(fn(Response $response) => decodeLines($response, Book::class) + ->evalTap(fn(Book $book) => io(fn() => $index->add($book))) + ->compile() + ->drain()) + ->unsafeRun(); +``` + +The server writes and flushes each chunk before producing the next, and the client reads the same way, so neither side holds more than a chunk of the export. See [Client](docs/core/client.md) and [Streaming](docs/core/streaming.md). + ## Documentation Full documentation is available in [docs/](docs/index.md). @@ -118,6 +136,7 @@ Full documentation is available in [docs/](docs/index.md). - [Core Concepts](docs/getting-started/core-concepts.md) - [Entity Decoding](docs/core/entity-decoding.md) - [Streaming](docs/core/streaming.md) +- [Client](docs/core/client.md) - [Middleware](docs/middleware/basics.md) - [REST API example](docs/examples/rest-api.md) - [API Reference](docs/api/functions.md) diff --git a/composer.json b/composer.json index e0f1724..f428ca0 100644 --- a/composer.json +++ b/composer.json @@ -13,7 +13,7 @@ "php": "^8.2 || ^8.3 || ^8.4", "phunkie/phunkie": "^1.5", "phunkie/effect": "^1.4", - "phunkie/streams": "^1.3" + "phunkie/streams": "^1.4" }, "require-dev": { "phpunit/phpunit": "^10.0 || ^11.0", diff --git a/composer.lock b/composer.lock index a9f571a..63d06bb 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "ee096c644f8d65cb4abc78c134d44aca", + "content-hash": "b23fd3bc0d826699c8b65212b260241a", "packages": [ { "name": "phunkie/effect", @@ -120,16 +120,16 @@ }, { "name": "phunkie/streams", - "version": "1.3.0", + "version": "1.4.0", "source": { "type": "git", "url": "https://github.com/phunkie/streams.git", - "reference": "e4cbb2c5bfd75b1ec6bee28d1c4789c0df259263" + "reference": "f9175ce2388fdd026b86f1384dd3aba63a132a83" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/phunkie/streams/zipball/e4cbb2c5bfd75b1ec6bee28d1c4789c0df259263", - "reference": "e4cbb2c5bfd75b1ec6bee28d1c4789c0df259263", + "url": "https://api.github.com/repos/phunkie/streams/zipball/f9175ce2388fdd026b86f1384dd3aba63a132a83", + "reference": "f9175ce2388fdd026b86f1384dd3aba63a132a83", "shasum": "" }, "require": { @@ -175,9 +175,9 @@ "description": "Functional Streams for PHP", "support": { "issues": "https://github.com/phunkie/streams/issues", - "source": "https://github.com/phunkie/streams/tree/1.3.0" + "source": "https://github.com/phunkie/streams/tree/1.4.0" }, - "time": "2026-09-26T18:21:12+00:00" + "time": "2026-09-27T10:28:50+00:00" } ], "packages-dev": [ diff --git a/docs/api/functions.md b/docs/api/functions.md index 0ffedcf..cb73b4c 100644 --- a/docs/api/functions.md +++ b/docs/api/functions.md @@ -4,8 +4,9 @@ Every helper lives under `Phunkie\Http4p\Functions`. Import what you use with `u | Namespace | Functions | |-----------|-----------| -| `Phunkie\Http4p\Functions` | `HttpRoutes`, `Request`, `Response`, `Headers`, `Status`, `StatusOk` and the other status constructors, `decode`, `FileResponse` | +| `Phunkie\Http4p\Functions` | `HttpRoutes`, `Request`, `Response`, `Headers`, `Status`, `StatusOk` and the other status constructors, `decode`, `decodeLines`, `FileResponse` | | `Phunkie\Http4p\Functions\routes` | `GET`, `POST`, `PUT`, `PATCH`, `DELETE` | +| `Phunkie\Http4p\Functions\client` | `send` | | `Phunkie\Http4p\Functions\response` | `Ok`, `Created`, `Accepted`, `NoContent`, `BadRequest`, `Unauthorized`, `Forbidden`, `NotFound`, `Conflict`, `UnprocessableEntity`, `InternalServerError` | | `Phunkie\Http4p\Functions\decoding` | `json` | | `Phunkie\Http4p\Functions\middleware` | `Through`, `Logger`, `Cors`, `Recover` | @@ -43,6 +44,15 @@ Every constructor takes the body first and an optional `EntityEncoder`, and retu - `UnprocessableEntity(mixed $body = null, ?EntityEncoder $encoder = null)` - `InternalServerError(mixed $body = null, ?EntityEncoder $encoder = null)` +## Decoding +- `decode(Request $request, string|callable|null $decoder = null, array $known = []): IO`, see [Entity Decoding](../core/entity-decoding.md) +- `decodeLines(Request|Response $message, string $class): Stream`, one instance of `$class` per line of newline-delimited JSON + +## Client +Namespace: `Phunkie\Http4p\Functions\client` + +- `send(Request $request, float $timeout = 30.0): IO`, the body a `Stream` read from the connection as it is compiled, see [Client](../core/client.md) + ## Streaming - `FileResponse(string $path): IO` diff --git a/docs/core/client.md b/docs/core/client.md new file mode 100644 index 0000000..236ec9d --- /dev/null +++ b/docs/core/client.md @@ -0,0 +1,70 @@ +# Client + +`send` sends a `Request` to another service and describes its response as an `IO`. A client has the same shape as a handler, `Request => IO`, so everything you know about responses applies: the status, the headers, and a body that is a `Stream` read from the connection chunk by chunk as it is compiled. + +```php +use Phunkie\Http4p\Method; +use Phunkie\Http4p\Response; + +use function Phunkie\Http4p\Functions\client\send; +use function Phunkie\Http4p\Functions\Headers; +use function Phunkie\Http4p\Functions\Request; + +$response = send(Request(Method::GET, 'http://catalogue.internal/books/1'))->unsafeRun(); + +$response->status->code; // 200 +$response->headers->get('content-type'); // application/json +implode('', $response->body->compile()->toArray()); // the whole body, when it fits in memory + +send(Request(Method::POST, 'http://catalogue.internal/books', Headers(['content-type' => 'application/json']), json_encode($book))); +``` + +The request goes out and the status and headers are read when the IO runs; the body is not read until the stream is compiled. A failed status is a response like any other: `send` does not throw on a 404 or a 500, only a `RuntimeException` when no connection could be made. The second argument is the timeout in seconds, for connecting and for each read. + +## Reading a large body + +The body is a stream of chunks, so it composes with every stream operation and is consumed in constant memory. `lines()` from phunkie/streams turns the chunks into lines however the chunks were cut: + +```php +use function Phunkie\Effect\Functions\io\io; + +send(Request(Method::GET, 'http://catalogue.internal/books/export')) + ->flatMap(fn(Response $response) => $response->body + ->lines() + ->evalTap(fn(string $line) => io(fn() => fwrite($file, $line . "\n"))) + ->compile() + ->drain()) + ->unsafeRun(); +``` + +## Decoding an export into entities + +`decodeLines` reads a body of newline-delimited JSON as a `Stream` of an entity, one per line, with the rules of [entity decoding](entity-decoding.md): every constructor parameter without a default is required, the `#[Generated]` ones included since the rows come from a store, value objects and enums are built from the scalars, and a line that does not fit fails the stream with a `DecodeFailure` when the stream reaches it. + +```php +use function Phunkie\Http4p\Functions\decodeLines; + +send(Request(Method::GET, 'http://catalogue.internal/books/export')) + ->flatMap(fn(Response $response) => decodeLines($response, Book::class) + ->evalTap(fn(Book $book) => io(fn() => $index->add($book))) + ->compile() + ->drain()) + ->unsafeRun(); +``` + +Between two http4p services, the exporting side streams `all(Book::class)->stream()` as lines (see [Streaming](streaming.md)) and the consuming side reads them back as `Book`s; neither holds more than a chunk of the export at a time. The integration test in `tests/Integration/ClientTest.php` sends 200,000 rows through this pair and checks that both processes stay flat. + +An arrow function captures variables by value. A closure that counts or collects by reference inside `evalTap` must therefore be created outside the `fn` that hands it over, or it writes to the arrow function's copy: + +```php +$count = 0; +$counting = function (Book $book) use (&$count) { + return io(function () use (&$count) { + $count++; + }); +}; + +send($request) + ->flatMap(fn(Response $response) => decodeLines($response, Book::class)->evalTap($counting)->compile()->drain()) + ->unsafeRun(); +``` diff --git a/docs/core/entity-decoding.md b/docs/core/entity-decoding.md index 39e32f9..7d77420 100644 --- a/docs/core/entity-decoding.md +++ b/docs/core/entity-decoding.md @@ -66,3 +66,7 @@ decode($req, Author::class) ->flatMap(fn(array $data) => create(Author::class, $data)->run($conn)) ->flatMap(Created(...)); ``` + +## Decoding a stream of lines + +`decodeLines($message, Author::class)` reads a request or response body of newline-delimited JSON as a `Stream` of `Author`, one per line, with the same rules and errors, and two differences that follow from the lines describing stored entities: every parameter without a default is required, as on `PUT`, and the `#[Generated]` parameters are expected too. Each line becomes an instance, not an array of fields, and a line that does not fit fails the stream with the `DecodeFailure` when the stream reaches it. See [Client](client.md) for reading another service's export this way. diff --git a/docs/core/streaming.md b/docs/core/streaming.md index 5d08902..cd16429 100644 --- a/docs/core/streaming.md +++ b/docs/core/streaming.md @@ -5,34 +5,52 @@ In **Http4p**, streaming is not an afterthought or a special mode: it is the def Every `Request` body is a Stream. Every `Response` body is a Stream. -This architecture ensures that your application uses constant memory, regardless of whether you are serving a 5KB JSON payload or a 5GB video file. +The server pulls one chunk at a time from the body, writes it and flushes it before pulling the next, and a client reads a response the same way, so memory holds one chunk whether the payload is 5KB of JSON or a 5GB export. ## Streaming Responses -When you create a response, you are providing a description of a stream of data. The server pulls from this stream and sends chunks to the client as they become available. +When you create a response, you are providing a description of a stream of data. `PhpServer` pulls from this stream and sends each chunk to the client as soon as it is produced: written, flushed through any output buffer and handed to the SAPI. No `Content-Length` is set for a streamed body, so the SAPI sends it with chunked transfer encoding. ### 1. Basic Output + Simple values are automatically converted to streams: + ```php -GET('/', fn() => Ok("Hello World")); // Streams: "Hello", " ", "World" (conceptually) +GET('/', fn() => Ok("Hello World")); ``` ### 2. Generating Data (Push) + You can stream data dynamically to the client. This is useful for large reports, logs, or real-time event feeds. `Stream()` is phunkie/streams' global factory, so it needs no import: ```php GET('/numbers', fn() => - // Create a stream that yields numbers 1 to 10 Ok(Stream(range(1, 10)) - ->map(fn($n) => "$n\n") // Format as lines - ->evalTap(fn() => io(fn() => usleep(100000))) // Simulate work/delay + ->map(fn($n) => "$n\n") + ->evalTap(fn() => io(fn() => usleep(100000))) ) ); ``` -In this example, the client receives each number as it is processed. The server does not wait for the loop to finish before sending the first byte. -### 3. File Streaming -For serving files, use the `FileResponse` helper which sets up an optimized stream from disk. +The client receives each number as it is produced: the server does not wait for the stream to end before sending the first byte. + +### 3. Exporting rows + +With [phunkie/phetch](https://github.com/phunkie/phetch) as the store, `stream()` on a query yields the rows one fetch at a time. Encode each as a line of newline-delimited JSON: + +```php +GET('/books/export', fn() => + all(Book::class)->stream()->run($conn) + ->flatMap(fn(Stream $books) => Ok($books->map(fn(Book $book) => json_encode($book) . "\n"))) + ->map(fn(Response $response) => $response->withHeader('content-type', 'application/x-ndjson')) +); +``` + +Nothing holds more than one row: it is fetched, encoded, written and flushed, and only then is the next one fetched. The [client](client.md) page shows the service that reads it back. + +### 4. File Streaming + +For serving files, use the `FileResponse` helper which sets up a stream from disk. ([Read more in File Streaming](../streaming/files.md)) ```php @@ -47,18 +65,25 @@ GET('/video', fn() => FileResponse('/media/movie.mp4')); Incoming request bodies are also Streams. This allows you to process uploads or large payloads chunk-by-chunk. ### 1. Process as you read -Instead of reading the whole body into memory with `$req->body->readAll()`, you can map over the stream. + +Instead of reading the whole body into memory, map over the stream: ```php POST('/uppercase', fn(Request $req) => - // Create a response that echoes the request body, transformed to uppercase Ok($req->body->map(fn($chunk) => strtoupper($chunk))) ); ``` -If you pipe a large file to this endpoint, it will stream the uppercase version back immediately, using negligible memory. + +Pipe a large file to this endpoint and it streams the uppercase version back as it reads, holding one chunk at a time. + +## Deployment + +`PhpServer` runs the application in whatever SAPI executes it. PHP's built-in server, `php -S`, is single-threaded: it serves one request at a time, so a long streamed response blocks every other client until it ends. Keep it for development. + +For production, run behind php-fpm with nginx or Apache in front, or under FrankenPHP or RoadRunner; they serve requests concurrently, send a body without `Content-Length` as chunked transfer encoding and forward every flush. Make sure the reverse proxy does not buffer responses (`proxy_buffering off;` in nginx for the streaming routes, or the `X-Accel-Buffering: no` header), or the client sees the whole body only at the end. ## Why this matters -1. **Time to First Byte (TTFB)**: Clients see data immediately. -2. **Memory Safety**: You never load 100MB of data into RAM to process it. -3. **Backpressure**: If the client is slow, `http4p` stops pulling from your source stream automatically. +1. **Time to first byte**: clients see data as soon as it is produced. +2. **Memory**: a body of any size is processed one chunk at a time. +3. **Backpressure**: a client that reads slowly blocks the write, which stops the pull from your source; see [Backpressure](../streaming/backpressure.md). diff --git a/docs/index.md b/docs/index.md index aaa6f0f..5528990 100644 --- a/docs/index.md +++ b/docs/index.md @@ -13,6 +13,7 @@ - [Entity Encoding](core/entity-encoding.md) - Converting values to streaming bodies - [Entity Decoding](core/entity-decoding.md) - Handling request bodies (Not Middleware!) - [Streaming](core/streaming.md) - Backpressure, cancellation, and memory efficiency +- [Client](core/client.md) - Calling another service and consuming its response as a stream ### Routing - [Route Definition](routing/definition.md) - Defining routes with `GET`, `POST`, etc. @@ -26,6 +27,7 @@ - [Database Streaming](streaming/database.md) - Streaming query results - [Custom Streams](streaming/custom.md) - Building custom streams - [Backpressure](streaming/backpressure.md) - Implicit flow control +- [Client](core/client.md) - Reading a streamed response, decoding an export line by line ### Middleware - [Basics](middleware/basics.md) - How usage works diff --git a/docs/streaming/backpressure.md b/docs/streaming/backpressure.md index c84eefe..1c0ec23 100644 --- a/docs/streaming/backpressure.md +++ b/docs/streaming/backpressure.md @@ -1,15 +1,16 @@ # Backpressure -Backpressure is the ability of a consumer to signal a producer to slow down. In `http4p` and `phunkie/streams`, backpressure is implicit and built-in via the **Pull-based** architecture. +Backpressure is the ability of a consumer to signal a producer to slow down. In `http4p` and `phunkie/streams` it is implicit: the pull-based pipeline reads nothing until the consumer asks for it, and the consumer is the socket. ## How it works -1. **Lazy Evaluation**: Streams are lazy. No data is read from the source (File, DB, Socket) until the consumer asks for it. -2. **Chunk-by-Chunk**: Data is processed in discrete chunks. -3. **Synchronous blocking**: In the standard PHP execution model, writing to the output buffer (`echo`) blocks if the buffer is full or the client is slow to receive. - - When `PhpServer` writes a chunk, `echo` blocks. +1. **Lazy evaluation**: a stream reads from its source (file, database statement, socket) only when compiled, one element at a time. +2. **One chunk in flight**: `PhpServer` pulls a chunk, writes it, flushes it, and only then pulls the next; nothing is buffered in between. +3. **The write blocks**: when the client is slow to receive, the flush blocks until the SAPI can send. - This blocks the `drain()` loop. - - This blocks the `pull()` from the source. - - Therefore, we stop reading from the source until the client is ready. + - This blocks the pull from the source. + - So the source is not read faster than the client takes the data. -This ensures that we never read more data into memory than what we can send, preventing Out-Of-Memory errors even when streaming gigabytes of data to a slow client. +The same holds on the client side: `send` gives a body that reads from the connection as it is compiled, so a producer that is slow to send slows the consumer down, and a consumer that processes slowly leaves the data in the socket. + +Memory stays at one chunk on both sides whatever the size of the body and the speed of either end. diff --git a/src/Client/ResponseHead.php b/src/Client/ResponseHead.php new file mode 100644 index 0000000..755afc9 --- /dev/null +++ b/src/Client/ResponseHead.php @@ -0,0 +1,58 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Http4p\Client; + +use Phunkie\Http4p\Headers; +use Phunkie\Http4p\Status; +use RuntimeException; + +use function Phunkie\Http4p\Functions\Headers; + +/** + * The status and headers of a response, read before any of its body. + */ +final readonly class ResponseHead +{ + public function __construct( + public Status $status, + public Headers $headers, + ) { + } + + /** + * From the lines PHP's http stream wrapper collects: a status line followed by header lines, + * repeated once per redirect, of which the last response is the one that matters. + * + * @param list $lines + */ + public static function fromWrapperData(array $lines): self + { + $start = null; + foreach ($lines as $index => $line) { + if (str_starts_with($line, 'HTTP/')) { + $start = $index; + } + } + if (null === $start) { + throw new RuntimeException('The response carries no HTTP status line.'); + } + + [, $code, $reason] = explode(' ', $lines[$start], 3) + [1 => '0', 2 => '']; + $headers = []; + foreach (array_slice($lines, $start + 1) as $line) { + [$name, $value] = explode(':', $line, 2) + [1 => '']; + $headers[strtolower(trim($name))] = trim($value); + } + + return new self(new Status((int) $code, $reason), Headers($headers)); + } +} diff --git a/src/Decoder/EntityDecoder.php b/src/Decoder/EntityDecoder.php index b43a3ed..a8ae8dd 100644 --- a/src/Decoder/EntityDecoder.php +++ b/src/Decoder/EntityDecoder.php @@ -43,6 +43,11 @@ final class EntityDecoder { private const GENERATED = 'Phunkie\Phetch\Attributes\Generated'; + private bool $includingGenerated = false; + + /** @var list|null */ + private ?array $parameters = null; + /** * @param class-string $class * @param array $known values the route already has, keyed by parameter name @@ -54,6 +59,20 @@ public function __construct( ) { } + /** + * A decoder for an entity as a store holds it: every parameter without a default is required, + * the generated ones included, since they were produced when the row was written. + * + * @param class-string $class + */ + public static function stored(string $class): self + { + $decoder = new self($class, Method::PUT); + $decoder->includingGenerated = true; + + return $decoder; + } + /** * @return array */ @@ -76,7 +95,7 @@ private function fields(array $json): array $fields = []; $errors = []; foreach ($this->parameters() as $param) { - if ($this->isGenerated($param)) { + if (! $this->includingGenerated && $this->isGenerated($param)) { continue; } @@ -130,7 +149,7 @@ private function fields(array $json): array */ private function parameters(): array { - return (new ReflectionClass($this->class))->getConstructor()?->getParameters() ?? []; + return $this->parameters ??= (new ReflectionClass($this->class))->getConstructor()?->getParameters() ?? []; } private function isGenerated(ReflectionParameter $param): bool diff --git a/src/Functions/client.php b/src/Functions/client.php index 76c4eaf..b1b7e55 100644 --- a/src/Functions/client.php +++ b/src/Functions/client.php @@ -9,5 +9,36 @@ * file that was distributed with this source code. */ -// HTTP Client functions - to be implemented +namespace Phunkie\Http4p\Functions\client { + use Phunkie\Effect\IO\IO; + use Phunkie\Http4p\Client\ResponseHead; + use Phunkie\Http4p\Request; + use Phunkie\Http4p\Response; + use Phunkie\Streams\IO\Network\HttpRequest; + + use function Phunkie\Effect\Functions\io\io; + + /** + * Send a request to another service and describe its response. + * + * The request goes out and the status and headers are read when the IO runs. The body is a + * Stream pulled from the connection chunk by chunk as it is compiled, so a large response is + * consumed in constant memory: compose it with lines(), decodeLines() or any stream operation. + * + * @param Request $request its uri absolute, http:// or https:// + * @param float $timeout seconds to wait for the connection and each read + * @return IO + */ + function send(Request $request, float $timeout = 30.0): IO + { + return io(function () use ($request, $timeout) { + $body = implode('', $request->body->compile()->toArray()); + $http = new HttpRequest($request->uri, $request->method->value, $request->headers->toArray(), '' === $body ? null : $body, $timeout); + $http->execute(); + $head = ResponseHead::fromWrapperData($http->getResponseHeaders()); + + return new Response($head->status, $head->headers, \Stream($http)); + }); + } +} diff --git a/src/Functions/decoding.php b/src/Functions/decoding.php index cceb632..b7b4a20 100644 --- a/src/Functions/decoding.php +++ b/src/Functions/decoding.php @@ -14,6 +14,8 @@ use Phunkie\Effect\IO\IO; use Phunkie\Http4p\Decoder\EntityDecoder; use Phunkie\Http4p\Request; + use Phunkie\Http4p\Response; + use Phunkie\Streams\Type\Stream; use function Phunkie\Effect\Functions\io\io; use function Phunkie\Http4p\Functions\decoding\json; @@ -41,6 +43,27 @@ function decode(Request $request, string|callable|null $decoder = null, array $k return io(fn () => $decoder(implode('', $request->body->compile()->toArray()))); } + + /** + * Decode a body of newline-delimited JSON, one entity per line, as a Stream of that entity. + * + * Each line is checked against the entity's constructor the way a full request body is, with + * the generated parameters expected as well since the rows come from a store, and becomes an + * instance. Blank lines are skipped. A line that does not fit fails the stream with a + * DecodeFailure when the stream reaches it, so nothing is decoded before it is needed. + * + * @param class-string $class + * @return Stream of instances of $class + */ + function decodeLines(Request|Response $message, string $class): Stream + { + $decoder = EntityDecoder::stored($class); + + return $message->body + ->lines() + ->filter(fn (string $line) => '' !== trim($line)) + ->map(fn (string $line) => new $class(...$decoder($line))); + } } namespace Phunkie\Http4p\Functions\decoding { diff --git a/src/Server/Output.php b/src/Server/Output.php new file mode 100644 index 0000000..7cc202a --- /dev/null +++ b/src/Server/Output.php @@ -0,0 +1,27 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Http4p\Server; + +/** + * Where a server writes a response: the status, the headers, then the body one chunk at a time. + */ +interface Output +{ + public function status(int $code): void; + + public function header(string $name, string $value): void; + + /** + * Write one chunk of the body and get it to the client before the next chunk is produced. + */ + public function write(string $chunk): void; +} diff --git a/src/Server/PhpServer.php b/src/Server/PhpServer.php index 356ef5a..7851bbb 100644 --- a/src/Server/PhpServer.php +++ b/src/Server/PhpServer.php @@ -13,7 +13,6 @@ use Phunkie\Effect\IO\IO; use Phunkie\Http4p\Encoder\JsonEncoder; -use Phunkie\Http4p\Headers; use Phunkie\Http4p\Method; use Phunkie\Http4p\Request; use Phunkie\Http4p\Response; @@ -29,10 +28,11 @@ use function Phunkie\Http4p\Functions\StatusInternalServerError; /** - * Simple HTTP server using PHP's built-in capabilities. + * Serves an application through PHP's own SAPI: one request per PHP process, as php-fpm, the + * built-in server, FrankenPHP or RoadRunner run it. * - * This is a basic implementation for development/testing. - * For production, use a proper server like ReactPHP or Swoole. + * The response body is written one chunk at a time and each chunk is flushed before the next one + * is produced, so a streamed body reaches the client as it is generated. */ final class PhpServer { @@ -41,10 +41,13 @@ final class PhpServer */ private $handler; + private Output $output; + /** * @param ImmList|callable(Request): IO $app + * @param Output|null $output where the response is written; the SAPI when not given */ - public function __construct(ImmList|callable $app) + public function __construct(ImmList|callable $app, ?Output $output = null) { if ($app instanceof ImmList) { $router = new Router($app); @@ -52,6 +55,7 @@ public function __construct(ImmList|callable $app) } else { $this->handler = $app; } + $this->output = $output ?? new SapiOutput(); } /** @@ -90,24 +94,21 @@ public function handleRequest(): IO } /** - * Send a response to the client. + * Send a response to the client: the status and headers, then the body chunk by chunk, each + * chunk written before the next one is pulled. * * @return IO Exit code */ public function sendResponse(Response $response): IO { return io(function () use ($response) { - // Set status - http_response_code($response->status->code); - - // Set headers + $this->output->status($response->status->code); foreach ($response->headers->toArray() as $name => $value) { - header("$name: $value"); + $this->output->header($name, $value); } })->flatMap( - fn() => - $response->body - ->evalTap(fn($chunk) => io(function () use ($chunk) { echo $chunk; })) + fn() => $response->body + ->evalTap(fn($chunk) => io(fn() => $this->output->write($chunk))) ->compile() ->drain() )->map(fn() => 0); diff --git a/src/Server/SapiOutput.php b/src/Server/SapiOutput.php new file mode 100644 index 0000000..014f8fb --- /dev/null +++ b/src/Server/SapiOutput.php @@ -0,0 +1,41 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Phunkie\Http4p\Server; + +/** + * Writes the response through PHP's SAPI: php-fpm, the built-in server, FrankenPHP, RoadRunner. + * + * Every chunk is flushed through any output buffer and the SAPI as soon as it is written, so a + * streamed body reaches the client as it is produced. No Content-Length is added; without one the + * SAPI sends the body chunked. + */ +final class SapiOutput implements Output +{ + public function status(int $code): void + { + http_response_code($code); + } + + public function header(string $name, string $value): void + { + header(sprintf('%s: %s', $name, $value)); + } + + public function write(string $chunk): void + { + echo $chunk; + if (ob_get_level() > 0) { + ob_flush(); + } + flush(); + } +} diff --git a/tests/Integration/ClientTest.php b/tests/Integration/ClientTest.php new file mode 100644 index 0000000..e74d21f --- /dev/null +++ b/tests/Integration/ClientTest.php @@ -0,0 +1,136 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Integration; + +use PHPUnit\Framework\TestCase; +use Phunkie\Http4p\Method; +use Phunkie\Http4p\Response; +use RuntimeException; +use Tests\Integration\Fixtures\Book; + +use function Phunkie\Effect\Functions\io\io; +use function Phunkie\Http4p\Functions\client\send; +use function Phunkie\Http4p\Functions\decodeLines; +use function Phunkie\Http4p\Functions\Headers; +use function Phunkie\Http4p\Functions\Request; + +final class ClientTest extends TestCase +{ + /** @var resource */ + private static $server; + + private static string $base; + + public static function setUpBeforeClass(): void + { + $port = self::freePort(); + self::$base = "http://127.0.0.1:$port"; + self::$server = proc_open( + [PHP_BINARY, '-d', 'xdebug.mode=off', '-S', "127.0.0.1:$port", __DIR__ . '/Fixtures/server.php'], + [1 => ['file', '/dev/null', 'w'], 2 => ['file', '/dev/null', 'w']], + $pipes, + ); + self::waitForPort($port); + } + + public static function tearDownAfterClass(): void + { + proc_terminate(self::$server); + proc_close(self::$server); + } + + public function testSendReadsTheStatusTheHeadersAndTheBody(): void + { + $response = send(Request(Method::GET, self::$base . '/hello'))->unsafeRun(); + + $this->assertSame(200, $response->status->code); + $this->assertSame('OK', $response->status->reason); + $this->assertSame('application/json', $response->headers->get('content-type')); + $this->assertSame('hello', implode('', $response->body->compile()->toArray())); + } + + public function testSendKeepsTheStatusOfAFailedResponse(): void + { + $response = send(Request(Method::GET, self::$base . '/missing'))->unsafeRun(); + + $this->assertSame(404, $response->status->code); + $this->assertSame('{"error":"nothing here"}', implode('', $response->body->compile()->toArray())); + } + + public function testSendCarriesTheRequestBody(): void + { + $request = Request(Method::POST, self::$base . '/echo', Headers(['content-type' => 'application/json']), '{"answer":42}'); + + $response = send($request)->unsafeRun(); + + $this->assertSame('{"answer":42}', implode('', $response->body->compile()->toArray())); + } + + public function testDecodeLinesTurnsAnExportIntoEntities(): void + { + $books = send(Request(Method::GET, self::$base . '/books?count=3')) + ->map(fn (Response $response) => decodeLines($response, Book::class)->compile()->toArray()) + ->unsafeRun(); + + $this->assertEquals([new Book(1, 'Book 1', 'Ada', false), new Book(2, 'Book 2', 'Ada', true), new Book(3, 'Book 3', 'Ada', false)], $books); + } + + public function testAnExportOfTwoHundredThousandRowsStaysFlatOnBothSides(): void + { + $count = 0; + $counting = function (Book $book) use (&$count) { + return io(function () use (&$count) { + $count++; + }); + }; + $before = memory_get_peak_usage(); + + send(Request(Method::GET, self::$base . '/books?count=200000')) + ->flatMap(fn (Response $response) => decodeLines($response, Book::class)->evalTap($counting)->compile()->drain()) + ->unsafeRun(); + + $this->assertSame(200000, $count); + $this->assertLessThan(8 * 1024 * 1024, memory_get_peak_usage() - $before, 'the client held more than a few chunks'); + $this->assertLessThan(32 * 1024 * 1024, $this->serverPeakMemory(), 'the server held the whole export'); + } + + private function serverPeakMemory(): int + { + $body = implode('', send(Request(Method::GET, self::$base . '/peak'))->unsafeRun()->body->compile()->toArray()); + + return json_decode($body, true, flags: JSON_THROW_ON_ERROR)['bytes']; + } + + private static function freePort(): int + { + $socket = stream_socket_server('tcp://127.0.0.1:0'); + $port = (int) substr(stream_socket_get_name($socket, false), strrpos(stream_socket_get_name($socket, false), ':') + 1); + fclose($socket); + + return $port; + } + + private static function waitForPort(int $port): void + { + for ($attempt = 0; $attempt < 50; $attempt++) { + $connection = @fsockopen('127.0.0.1', $port, $errno, $error, 0.2); + if (false !== $connection) { + fclose($connection); + + return; + } + usleep(100000); + } + + throw new RuntimeException(sprintf('The fixture server did not come up on port %d.', $port)); + } +} diff --git a/tests/Integration/Fixtures/Book.php b/tests/Integration/Fixtures/Book.php new file mode 100644 index 0000000..d180771 --- /dev/null +++ b/tests/Integration/Fixtures/Book.php @@ -0,0 +1,26 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Integration\Fixtures; + +use Phunkie\Phetch\Attributes\Generated; + +final readonly class Book +{ + public function __construct( + #[Generated] + public int $id, + public string $title, + public string $author, + public bool $inStock, + ) { + } +} diff --git a/tests/Integration/Fixtures/server.php b/tests/Integration/Fixtures/server.php new file mode 100644 index 0000000..80df467 --- /dev/null +++ b/tests/Integration/Fixtures/server.php @@ -0,0 +1,53 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +use Phunkie\Http4p\Request; +use Phunkie\Http4p\Response; +use Phunkie\Http4p\Server\PhpServer; +use Phunkie\Streams\IO\Resource; +use Tests\Integration\Fixtures\Book; + +use function Phunkie\Http4p\Functions\decode; +use function Phunkie\Http4p\Functions\HttpRoutes; +use function Phunkie\Http4p\Functions\response\NotFound; +use function Phunkie\Http4p\Functions\response\Ok; +use function Phunkie\Http4p\Functions\routes\GET; +use function Phunkie\Http4p\Functions\routes\POST; + +require dirname(__DIR__, 3) . '/vendor/autoload.php'; + +$books = fn (int $count) => new class ($count) implements Resource { + private int $next = 1; + + public function __construct(private int $count) + { + } + + public function pull(int $chunkSize): mixed + { + if ($this->next > $this->count) { + return Resource::EOF; + } + $book = new Book($this->next, "Book {$this->next}", 'Ada', 0 === $this->next % 2); + $this->next++; + + return json_encode($book) . "\n"; + } +}; + +(new PhpServer(HttpRoutes( + GET('/hello', fn () => Ok('hello')), + GET('/missing', fn () => NotFound(['error' => 'nothing here'])), + POST('/echo', fn (Request $req) => decode($req)->flatMap(fn ($json) => Ok($json))), + GET('/books', fn (Request $req) => Ok(\Stream($books((int) ($req->query('count') ?? 3)))) + ->map(fn (Response $response) => $response->withHeader('content-type', 'application/x-ndjson'))), + GET('/peak', fn () => Ok(['bytes' => memory_get_peak_usage()])), +)))->run()->unsafeRun(); diff --git a/tests/Unit/DecodeLinesTest.php b/tests/Unit/DecodeLinesTest.php new file mode 100644 index 0000000..2461171 --- /dev/null +++ b/tests/Unit/DecodeLinesTest.php @@ -0,0 +1,61 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit; + +use PHPUnit\Framework\TestCase; +use Phunkie\Http4p\DecodeFailure; +use Tests\Integration\Fixtures\Book; + +use function Phunkie\Http4p\Functions\decodeLines; +use function Phunkie\Http4p\Functions\response\Ok; + +final class DecodeLinesTest extends TestCase +{ + public function testDecodesOneEntityPerLineWhateverTheChunkBoundaries(): void + { + $response = Ok(\Stream('{"id":1,"title":"Notes","author":"Ada","inStock":true}' . "\n" . '{"id":2,"ti', 'tle":"Sketch","author":"Ada","inStock":false}' . "\n"))->unsafeRun(); + + $books = decodeLines($response, Book::class)->compile()->toArray(); + + $this->assertEquals([new Book(1, 'Notes', 'Ada', true), new Book(2, 'Sketch', 'Ada', false)], $books); + } + + public function testSkipsBlankLinesAndAcceptsAMissingFinalNewline(): void + { + $response = Ok(\Stream("\n" . '{"id":1,"title":"Notes","author":"Ada","inStock":true}' . "\n\n" . '{"id":2,"title":"Sketch","author":"Ada","inStock":false}'))->unsafeRun(); + + $this->assertCount(2, decodeLines($response, Book::class)->compile()->toArray()); + } + + public function testFailsTheStreamWhenItReachesTheLineThatDoesNotFit(): void + { + $response = Ok(\Stream('{"id":1,"title":"Notes","author":"Ada","inStock":true}' . "\n", '{"id":"two","title":"Sketch"}' . "\n"))->unsafeRun(); + $seen = []; + + try { + decodeLines($response, Book::class) + ->evalTap(function (Book $book) use (&$seen) { + return \Phunkie\Effect\Functions\io\io(function () use (&$seen, $book) { + $seen[] = $book->id; + }); + }) + ->compile() + ->drain() + ->unsafeRun(); + $this->fail('Expected a DecodeFailure.'); + } catch (DecodeFailure $e) { + $this->assertSame(['id' => 'expected int, got string', 'author' => 'missing', 'inStock' => 'missing'], $e->errors()); + } + + $this->assertSame([1], $seen); + } +} diff --git a/tests/Unit/PhpServerTest.php b/tests/Unit/PhpServerTest.php index c68f8d0..f0fb5be 100644 --- a/tests/Unit/PhpServerTest.php +++ b/tests/Unit/PhpServerTest.php @@ -14,7 +14,9 @@ use PHPUnit\Framework\TestCase; use Phunkie\Http4p\Request; use Phunkie\Http4p\Response; +use Phunkie\Http4p\Server\Output; use Phunkie\Http4p\Server\PhpServer; +use Phunkie\Streams\IO\Resource; use RuntimeException; use function Phunkie\Effect\Functions\io\io; @@ -64,14 +66,70 @@ public function testHandleRequestTurnsAThrowingHandlerIntoA500Response(): void $this->assertStringContainsString('boom', implode('', $response->body->toArray())); } - public function testSendResponseWritesTheBody(): void + public function testSendResponseWritesTheStatusTheHeadersAndTheBody(): void { - $server = new PhpServer(HttpRoutes()); + $output = $this->recordingOutput(); + $server = new PhpServer(HttpRoutes(), $output); - ob_start(); $server->sendResponse(Ok('hello')->unsafeRun())->unsafeRun(); - $output = ob_get_clean(); - $this->assertSame('hello', $output); + $this->assertSame(['status 200', 'header content-type: application/json', 'write hello'], $output->log); + } + + public function testSendResponseWritesEachChunkBeforePullingTheNext(): void + { + $output = $this->recordingOutput(); + $source = new class ($output->log) implements Resource { + private array $chunks = ['one', 'two']; + + public function __construct(private array &$log) + { + } + + public function pull(int $chunkSize): mixed + { + $chunk = array_shift($this->chunks); + $this->log[] = null === $chunk ? 'pull end' : "pull $chunk"; + + return $chunk ?? Resource::EOF; + } + }; + $server = new PhpServer(HttpRoutes(), $output); + + $server->sendResponse(Ok(\Stream($source))->unsafeRun())->unsafeRun(); + + $this->assertSame(['status 200', 'pull one', 'write one', 'pull two', 'write two', 'pull end'], $output->log); + } + + public function testSendResponseAddsNoContentLengthToAStreamedBody(): void + { + $output = $this->recordingOutput(); + $server = new PhpServer(HttpRoutes(), $output); + + $server->sendResponse(Ok(\Stream('a', 'b'))->unsafeRun())->unsafeRun(); + + $this->assertSame(['status 200', 'write a', 'write b'], $output->log); + } + + private function recordingOutput(): Output + { + return new class () implements Output { + public array $log = []; + + public function status(int $code): void + { + $this->log[] = "status $code"; + } + + public function header(string $name, string $value): void + { + $this->log[] = "header $name: $value"; + } + + public function write(string $chunk): void + { + $this->log[] = "write $chunk"; + } + }; } } diff --git a/tests/Unit/ResponseHeadTest.php b/tests/Unit/ResponseHeadTest.php new file mode 100644 index 0000000..1945def --- /dev/null +++ b/tests/Unit/ResponseHeadTest.php @@ -0,0 +1,51 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Unit; + +use PHPUnit\Framework\TestCase; +use Phunkie\Http4p\Client\ResponseHead; +use RuntimeException; + +final class ResponseHeadTest extends TestCase +{ + public function testReadsTheStatusAndTheHeadersWithLowercaseNames(): void + { + $head = ResponseHead::fromWrapperData(['HTTP/1.1 404 Not Found', 'Content-Type: application/json', 'X-Request-Id: abc ']); + + $this->assertSame(404, $head->status->code); + $this->assertSame('Not Found', $head->status->reason); + $this->assertSame(['content-type' => 'application/json', 'x-request-id' => 'abc'], $head->headers->toArray()); + } + + public function testKeepsTheLastResponseAfterRedirects(): void + { + $head = ResponseHead::fromWrapperData(['HTTP/1.1 302 Found', 'Location: /there', 'HTTP/1.1 200 OK', 'Content-Length: 2']); + + $this->assertSame(200, $head->status->code); + $this->assertSame(['content-length' => '2'], $head->headers->toArray()); + } + + public function testAcceptsAStatusLineWithoutAReason(): void + { + $head = ResponseHead::fromWrapperData(['HTTP/1.1 204']); + + $this->assertSame(204, $head->status->code); + $this->assertSame('', $head->status->reason); + } + + public function testRefusesAResponseWithoutAStatusLine(): void + { + $this->expectException(RuntimeException::class); + + ResponseHead::fromWrapperData(['Content-Type: text/plain']); + } +}