Developing 1.1.0

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<Response>` 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)

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",

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<mixed>`, 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<Response>`, the body a `Stream` read from the connection as it is compiled, see [Client](../core/client.md)
+
 ## Streaming
 - `FileResponse(string $path): IO<Response>`
 

docs/core/client.md

@@ -0,0 +1,70 @@
+# Client
+
+`send` sends a `Request` to another service and describes its response as an `IO<Response>`. A client has the same shape as a handler, `Request => IO<Response>`, 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();
+```

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.

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).

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

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.

src/Client/ResponseHead.php

@@ -0,0 +1,58 @@
+<?php
+
+/*
+ * This file is part of Phunkie Http4p.
+ *
+ * (c) Marcello Duarte <marcello.duarte@gmail.com>
+ *
+ * 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<string> $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));
+    }
+}