diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 93efbf6..7accca7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -83,7 +83,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, windows-latest, macos-latest] - php: ['8.3', '8.4'] + php: ['8.3', '8.4', '8.5'] stability: [prefer-stable] include: - os: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 8afc23d..75994ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,23 @@ # Release Notes -## [Unreleased](https://github.com/Thavarshan/fetch-php/compare/3.5.1...HEAD) +## [Unreleased](https://github.com/Thavarshan/fetch-php/compare/3.6.0...HEAD) + +## [v3.6.0](https://github.com/Thavarshan/fetch-php/compare/3.5.1...3.6.0) - 2026-07-21 + +### Added + +- **Streaming responses and Server-Sent Events** for consuming response bodies incrementally instead of buffering them into memory: + - `Fetch\Http\StreamedResponse` — an unbuffered response returned by the new `stream()` methods; pull raw chunks via `stream()`, newline-delimited lines via `lines()`, iterate the object directly, or fall back to a buffered `Response` with `buffer()`. + - `Fetch\Http\EventSource` — a WHATWG-compliant `text/event-stream` parser that lazily yields events and tracks `lastEventId()`/`reconnectionTime()`. + - `Fetch\Http\ServerSentEvent` — an immutable event value object with `data`, `type`, `id`, and `retry` fields plus `json()` and `isDone()` helpers. + - `stream()` and `sse()` methods on `ClientHandler` and `Client`, exposed through the `RequestExecutor` interface. + - `fetch_stream()` and `fetch_sse()` global helper functions mirroring the existing `fetch()`/`get()` style. + - `ContentType::EVENT_STREAM` and `ContentType::NDJSON` cases with `isEventStream()`/`isStreamable()` helpers. +- Added PHP 8.5 to the CI test matrix across Ubuntu, Windows, and macOS. + +### Changed + +- Streaming requests are synchronous and intentionally bypass the response cache, consistent with the library's existing sync-only caching behaviour. ## [v3.5.1](https://github.com/Thavarshan/fetch-php/compare/3.5.0...3.5.1) - 2026-06-06 diff --git a/CODE_MAP.md b/CODE_MAP.md index 7184b52..02bab88 100644 --- a/CODE_MAP.md +++ b/CODE_MAP.md @@ -15,6 +15,9 @@ This reference captures the concrete public surface of the FetchPHP library as i - Maintains a static `Client` instance; `reset=true` recreates it. - Passing `$options` clones the current handler with merged defaults (exceptions are wrapped in `RuntimeException` with context). - Verb helpers `get/post/put/patch/delete()` call `request_method()` which coerces array bodies to JSON unless `$dataIsQuery` is true. +- Streaming helpers: + - `fetch_stream(string $url, ?array $options = []): Fetch\Interfaces\StreamedResponse` – returns an unbuffered `StreamedResponse`; the body is pulled incrementally via `stream()`/`lines()` rather than read into memory. Method defaults to GET (override with `$options['method']`). + - `fetch_sse(string $url, ?array $options = []): Fetch\Http\EventSource` – consumes a `text/event-stream` response as lazily-yielded `Fetch\Http\ServerSentEvent` objects; adds an `Accept: text/event-stream` header automatically. - Matrix async bridge helpers (`async`, `await`, `all`, `race`, `map`, `batch`, `retry`) are re-exported when the corresponding `\Matrix\*` functions exist. - Internal helpers: - `process_request_options(array $options)` normalizes method enums/strings, headers, body precedence (`json` > `form` > `multipart` > `body`) and high-level flags. @@ -96,6 +99,20 @@ Key runtime behaviors: - Debugging hooks: `withDebugInfo(DebugInfo $info)` and `getDebugInfo()` to inspect per-request snapshots. - `ResponseImmutabilityTrait` keeps buffered body contents in sync when streams are replaced. +### `Fetch\Http\StreamedResponse` + +- Extends `GuzzleHttp\Psr7\Response`, implements `Fetch\Interfaces\StreamedResponse` + `IteratorAggregate`. +- Returned by `ClientHandler::stream()` / `Client::stream()` / `fetch_stream()` when a request runs with `stream => true`. The body is **not** buffered. +- Consumption: `stream(int $chunkSize = 8192)` yields raw chunks; `lines()` yields newline-delimited lines (handles `\r\n` and chunk boundaries); iterating the object itself yields chunks. +- `sse()` wraps the stream in an `EventSource`; `buffer()` drains the remaining stream into a normal `Fetch\Http\Response`. +- Mirrors `Response` status/header helpers: `status()`, `statusEnum()`, `ok()`, `failed()`, `headers()`, `header()`, `contentType()`, `contentTypeEnum()`. + +### `Fetch\Http\EventSource` & `Fetch\Http\ServerSentEvent` + +- `EventSource` (implements `IteratorAggregate`) parses a `text/event-stream` body per the WHATWG SSE algorithm, yielding events lazily via `events()` (or direct iteration). Tracks `lastEventId()` and `reconnectionTime()` across the stream. +- `ServerSentEvent` – immutable DTO with `data`, `type`, `id`, `retry`; helpers `json()` (decode payload), `isDone()` (detects the `[DONE]` sentinel), and `__toString()`. +- Returned by `ClientHandler::sse()` / `Client::sse()` / `fetch_sse()`. Streaming is synchronous and bypasses the response cache. + --- ## Support Services diff --git a/README.md b/README.md index a5f2da4..384b275 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ Full documentation can be found [here](https://fetch-php.thavarshan.com/) - **Promise-based API**: Use familiar `.then()`, `.catch()`, and `.finally()` methods for async operations - **Fluent Interface**: Build requests with a clean, chainable API - **Built on Guzzle**: Benefit from Guzzle's robust functionality with a more elegant API +- **Streaming & Server-Sent Events**: Consume response bodies incrementally (`response.body`-style) and parse `text/event-stream` responses — ideal for streaming LLM APIs and live feeds - **Retry Mechanics**: Configurable retry logic with exponential backoff for transient failures - **RFC 7234 HTTP Caching**: Full caching support with ETag/Last-Modified revalidation, stale-while-revalidate, and stale-if-error - **Connection Pooling**: Reuse TCP connections across requests with global connection pool and DNS caching diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index 0790adb..3276ce2 100644 --- a/docs/.vitepress/config.mts +++ b/docs/.vitepress/config.mts @@ -207,6 +207,10 @@ export default defineConfig({ text: "Connection Pooling & HTTP/2", link: "/guide/connection-pooling", }, + { + text: "Streaming & Server-Sent Events", + link: "/guide/streaming", + }, { text: "File Uploads", link: "/guide/file-uploads" }, { text: "Custom Clients", diff --git a/docs/guide/streaming.md b/docs/guide/streaming.md new file mode 100644 index 0000000..3cf2e73 --- /dev/null +++ b/docs/guide/streaming.md @@ -0,0 +1,128 @@ +--- +title: Streaming & Server-Sent Events +description: Consume response bodies incrementally and parse text/event-stream responses for streaming LLM APIs and live feeds. +--- + +# Streaming & Server-Sent Events + +Most responses in Fetch PHP are fully buffered—the body is read into memory and exposed through helpers like `json()` and `text()`. That is the right default for typical API calls, but it breaks down for large downloads and for endpoints that emit data over time, such as streaming LLM completions. + +For those cases Fetch PHP provides an unbuffered path that mirrors JavaScript's `response.body`: a [`StreamedResponse`](#streamedresponse) you pull chunks or lines from as they arrive, plus a first-class [Server-Sent Events](#server-sent-events) consumer. + +> Streaming is **synchronous** and **bypasses the response cache**. The body is never read into memory up front, so a streamed response can only be traversed once—use `buffer()` if you need random access. + +## Streaming a response body + +Use the `fetch_stream()` helper, or `->stream()` on a client/handler. Nothing is read until you iterate. + +```php +use function fetch_stream; + +$response = fetch_stream('https://example.com/large-export.csv'); + +// Pull raw chunks (default 8 KB) as they arrive. +foreach ($response->stream() as $chunk) { + echo $chunk; +} +``` + +Prefer `lines()` for text protocols—it handles `\n`/`\r\n` and reassembles lines split across chunk boundaries: + +```php +foreach (fetch_stream('https://example.com/events.log')->lines() as $line) { + process($line); +} +``` + +The fluent API works too, so you can attach headers, auth, or a base URI first: + +```php +use Fetch\Enum\Method; + +$response = fetch_client() + ->getHandler() + ->withToken($apiKey) + ->stream(Method::GET, 'https://example.com/feed'); + +if ($response->ok()) { + foreach ($response->lines() as $line) { + // ... + } +} +``` + +### Falling back to a buffered response + +Call `buffer()` to drain the remaining stream into a regular `Fetch\Http\Response`: + +```php +$response = fetch_stream('https://api.example.com/report'); + +$data = $response->buffer()->json(); +``` + +## Server-Sent Events + +`fetch_sse()` (or `->sse()`) returns an `EventSource` that lazily yields `ServerSentEvent` objects parsed according to the WHATWG Server-Sent Events specification. An `Accept: text/event-stream` header is added automatically. + +```php +use function fetch_sse; + +$events = fetch_sse('https://api.example.com/v1/stream'); + +foreach ($events as $event) { + if ($event->isDone()) { // detects the "[DONE]" sentinel many APIs send + break; + } + + $payload = $event->json(); // decode the data: field as JSON + echo $payload['delta'] ?? ''; +} +``` + +### Streaming an LLM completion + +SSE endpoints are frequently `POST` requests. Pass a method and body through the options array: + +```php +$events = fetch_sse('https://api.example.com/v1/chat/completions', [ + 'method' => 'POST', + 'headers' => ['Authorization' => "Bearer {$apiKey}"], + 'json' => [ + 'model' => 'example-model', + 'stream' => true, + 'messages' => [ + ['role' => 'user', 'content' => 'Write a haiku about PHP.'], + ], + ], +]); + +foreach ($events as $event) { + if ($event->isDone()) { + break; + } + + $delta = $event->json()['choices'][0]['delta']['content'] ?? ''; + echo $delta; +} +``` + +### The `ServerSentEvent` object + +Each dispatched event exposes the parsed fields: + +| Property / method | Description | +| ----------------- | ----------- | +| `data` | The event payload (multiple `data:` lines are joined with `\n`). | +| `type` | The event type from the `event:` field (defaults to `"message"`). | +| `id` | The last event ID from the `id:` field, if any. | +| `retry` | The reconnection time in milliseconds from the `retry:` field, if any. | +| `json()` | Decode `data` as JSON. | +| `isDone()` | Whether `data` is the conventional `[DONE]` termination sentinel. | + +The `EventSource` also tracks stream-level state via `lastEventId()` and `reconnectionTime()`. + +## When to stream + +- **Stream** for large downloads, long-lived feeds, and any `text/event-stream` endpoint (streaming AI/LLM APIs, live dashboards, log tails). +- **Buffer** (the default `fetch()`/`get()`/`post()` helpers) for ordinary JSON APIs where you want the whole response at once and features like caching and retries. diff --git a/src/Fetch/Concerns/PerformsHttpRequests.php b/src/Fetch/Concerns/PerformsHttpRequests.php index 02474c0..6ac1963 100644 --- a/src/Fetch/Concerns/PerformsHttpRequests.php +++ b/src/Fetch/Concerns/PerformsHttpRequests.php @@ -8,8 +8,11 @@ use Fetch\Enum\ContentType; use Fetch\Enum\Method; use Fetch\Exceptions\RequestException as FetchRequestException; +use Fetch\Http\EventSource; use Fetch\Http\Response; +use Fetch\Http\StreamedResponse; use Fetch\Interfaces\Response as ResponseInterface; +use Fetch\Interfaces\StreamedResponse as StreamedResponseInterface; use Fetch\Support\RequestContext; use Fetch\Support\RequestOptions; use GuzzleHttp\Exception\GuzzleException; @@ -145,6 +148,57 @@ public function options(string $uri): ResponseInterface|PromiseInterface return $this->sendRequest(Method::OPTIONS, $uri); } + /** + * Send a request and return an unbuffered, streamable response. + * + * The response body is not read into memory; callers pull chunks or lines + * as the server produces them (see {@see StreamedResponse}). Streaming is + * always synchronous and bypasses the response cache. + * + * @param array $options Additional options + */ + public function stream( + Method|string $method = Method::GET, + string $uri = '', + array $options = [], + ): StreamedResponseInterface { + $options['stream'] = true; + + return $this->sendStreamingRequest($method, $uri, $options); + } + + /** + * Send a request and consume the response as Server-Sent Events. + * + * Adds an `Accept: text/event-stream` header when none is present and + * returns an {@see EventSource} yielding parsed {@see ServerSentEvent}s. + * + * @param array $options Additional options + */ + public function sse( + Method|string $method = Method::GET, + string $uri = '', + array $options = [], + ): EventSource { + $headers = $options['headers'] ?? []; + + $hasAccept = false; + foreach (array_keys($headers) as $name) { + if (strcasecmp((string) $name, 'Accept') === 0) { + $hasAccept = true; + break; + } + } + + if (! $hasAccept) { + $headers['Accept'] = ContentType::EVENT_STREAM->value; + } + + $options['headers'] = $headers; + + return new EventSource($this->stream($method, $uri, $options)); + } + /** * Send an HTTP request. * @@ -339,6 +393,52 @@ public function getEffectiveTimeout(?RequestContext $context = null): int return self::DEFAULT_TIMEOUT; } + /** + * Execute a synchronous streaming request and wrap the raw PSR-7 response. + * + * @param array $options Additional options + */ + protected function sendStreamingRequest( + Method|string $method, + string $uri, + array $options = [], + ): StreamedResponseInterface { + $methodStr = $method instanceof Method ? $method->value : strtoupper($method); + + // Streaming is inherently synchronous: force async off for this path. + $requestOptions = RequestOptions::merge( + $this->options, + $options, + ['method' => $methodStr, 'uri' => $uri, 'async' => false, 'stream' => true], + ); + + RequestOptions::validate($requestOptions); + + $context = RequestContext::fromOptions($requestOptions); + $fullUri = $this->buildFullUriFromContext($context); + $guzzleOptions = $context->toGuzzleOptions(); + $guzzleOptions['stream'] = true; + + $requestId = $this->startProfiling($methodStr, $fullUri); + + if (method_exists($this, 'logRequest')) { + $this->logRequest($methodStr, $fullUri, $guzzleOptions); + } + + try { + $this->recordProfilingEvent($requestId, 'request_sent'); + + $psrResponse = $this->getHttpClient()->request($methodStr, $fullUri, $guzzleOptions); + + $this->recordProfilingEvent($requestId, 'response_start'); + $this->endProfiling($requestId, $psrResponse->getStatusCode()); + + return StreamedResponse::createFromBase($psrResponse); + } catch (\Throwable $e) { + throw $this->withErrorContext($e, $methodStr, $fullUri); + } + } + /** * Apply body-related options without mutating handler state. * diff --git a/src/Fetch/Enum/ContentType.php b/src/Fetch/Enum/ContentType.php index 7c43fe5..e6f63f4 100644 --- a/src/Fetch/Enum/ContentType.php +++ b/src/Fetch/Enum/ContentType.php @@ -19,6 +19,8 @@ enum ContentType: string case ZIP = 'application/zip'; case JAVASCRIPT = 'application/javascript'; case CSS = 'text/css'; + case EVENT_STREAM = 'text/event-stream'; + case NDJSON = 'application/x-ndjson'; /** * Get a content type from a string. @@ -85,11 +87,31 @@ public function isText(): bool { return match ($this) { // These are text-based content types - self::JSON, self::FORM_URLENCODED, self::TEXT, self::HTML, self::XML, self::CSV => true, + self::JSON, self::FORM_URLENCODED, self::TEXT, self::HTML, self::XML, self::CSV, self::EVENT_STREAM, self::NDJSON => true, // These are binary/non-text content types self::MULTIPART => false, // Default for any new enum values added in the future default => false, }; } + + /** + * Check if the content type is a Server-Sent Events stream. + */ + public function isEventStream(): bool + { + return $this === self::EVENT_STREAM; + } + + /** + * Check if the content type is typically consumed as an incremental + * stream rather than a single buffered payload (e.g. SSE or NDJSON). + */ + public function isStreamable(): bool + { + return match ($this) { + self::EVENT_STREAM, self::NDJSON => true, + default => false, + }; + } } diff --git a/src/Fetch/Http/Client.php b/src/Fetch/Http/Client.php index c97da36..a54fbd0 100644 --- a/src/Fetch/Http/Client.php +++ b/src/Fetch/Http/Client.php @@ -11,6 +11,7 @@ use Fetch\Exceptions\RequestException; use Fetch\Interfaces\ClientHandler as ClientHandlerInterface; use Fetch\Interfaces\Response as ResponseInterface; +use Fetch\Interfaces\StreamedResponse as StreamedResponseInterface; use GuzzleHttp\ClientInterface as GuzzleClientInterface; use GuzzleHttp\Exception\ConnectException; use GuzzleHttp\Exception\RequestException as GuzzleRequestException; @@ -362,6 +363,36 @@ public function options(string $url, ?array $options = []): ResponseInterface return $this->methodRequest(Method::OPTIONS, $url, null, ContentType::JSON, $options); } + /** + * Send a request and return an unbuffered, streamable response. + * + * @param string $url The URL to fetch + * @param array|null $options Request options + * @param string|Method $method The HTTP method + */ + public function stream( + string $url, + ?array $options = [], + string|Method $method = Method::GET, + ): StreamedResponseInterface { + return $this->handler->withOptions($options ?? [])->stream($method, $url); + } + + /** + * Send a request and consume the response as Server-Sent Events. + * + * @param string $url The URL to fetch + * @param array|null $options Request options + * @param string|Method $method The HTTP method + */ + public function sse( + string $url, + ?array $options = [], + string|Method $method = Method::GET, + ): EventSource { + return $this->handler->withOptions($options ?? [])->sse($method, $url); + } + /** * Get the underlying Guzzle HTTP client. */ diff --git a/src/Fetch/Http/EventSource.php b/src/Fetch/Http/EventSource.php new file mode 100644 index 0000000..308312d --- /dev/null +++ b/src/Fetch/Http/EventSource.php @@ -0,0 +1,215 @@ +isDone()) { + * break; + * } + * + * $payload = $event->json(); + * } + * ``` + * + * @see https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation + * + * @implements IteratorAggregate + */ +class EventSource implements IteratorAggregate +{ + /** + * The last event ID seen on the stream. + * + * Per the specification this persists across events and is carried onto + * subsequent events that omit an `id:` field. + */ + protected ?string $lastEventId = null; + + /** + * The most recent reconnection time (ms) advertised via a `retry:` field. + */ + protected ?int $reconnectionTime = null; + + public function __construct( + protected readonly StreamedResponseInterface $response, + ) {} + + /** + * The underlying streamed response. + */ + public function response(): StreamedResponseInterface + { + return $this->response; + } + + /** + * The last event ID observed on the stream so far. + */ + public function lastEventId(): ?string + { + return $this->lastEventId; + } + + /** + * The most recent reconnection time (in milliseconds), if advertised. + */ + public function reconnectionTime(): ?int + { + return $this->reconnectionTime; + } + + /** + * Yield parsed events as they arrive on the stream. + * + * @return Generator + */ + public function events(): Generator + { + // Per-event accumulators, reset after each dispatch. + $dataBuffer = ''; + $eventType = ''; + $hasData = false; + $hasFields = false; + + foreach ($this->response->lines() as $line) { + // A blank line dispatches the buffered event (if any). + if ($line === '') { + if ($hasData || $hasFields) { + $event = $this->buildEvent($dataBuffer, $eventType, $hasData); + + if ($event !== null) { + yield $event; + } + } + + $dataBuffer = ''; + $eventType = ''; + $hasData = false; + $hasFields = false; + + continue; + } + + // Lines beginning with a colon are comments and are ignored. + if ($line[0] === ':') { + continue; + } + + [$field, $value] = $this->splitField($line); + + switch ($field) { + case 'event': + $eventType = $value; + $hasFields = true; + break; + + case 'data': + $dataBuffer .= $value."\n"; + $hasData = true; + $hasFields = true; + break; + + case 'id': + // The spec ignores an id containing a NUL character. + if (! str_contains($value, "\0")) { + $this->lastEventId = $value; + } + $hasFields = true; + break; + + case 'retry': + if ($value !== '' && ctype_digit($value)) { + $this->reconnectionTime = (int) $value; + } + $hasFields = true; + break; + + default: + // Unknown fields are ignored per the specification. + break; + } + } + + // The stream ended without a trailing blank line: the spec discards + // any incomplete event, so no final dispatch is performed here. + } + + /** + * Alias for {@see events()} so the source is directly foreach-able. + * + * @return Generator + */ + public function getIterator(): Generator + { + yield from $this->events(); + } + + /** + * Build a dispatchable event from the current accumulators. + * + * Returns null when the buffer holds no data lines (only metadata), which + * per the spec must not fire a message event. + */ + protected function buildEvent(string $dataBuffer, string $eventType, bool $hasData): ?ServerSentEvent + { + if (! $hasData) { + return null; + } + + // A single trailing newline is stripped from the data buffer. + $data = substr($dataBuffer, -1) === "\n" + ? substr($dataBuffer, 0, -1) + : $dataBuffer; + + return new ServerSentEvent( + data: $data, + type: $eventType !== '' ? $eventType : 'message', + id: $this->lastEventId, + retry: $this->reconnectionTime, + ); + } + + /** + * Split a field line into its name and value per the SSE grammar. + * + * A line with no colon is a field with an empty value. A single space + * immediately following the colon is stripped from the value. + * + * @return array{0: string, 1: string} + */ + protected function splitField(string $line): array + { + $pos = strpos($line, ':'); + + if ($pos === false) { + return [$line, '']; + } + + $field = substr($line, 0, $pos); + $value = substr($line, $pos + 1); + + if ($value !== '' && $value[0] === ' ') { + $value = substr($value, 1); + } + + return [$field, $value]; + } +} diff --git a/src/Fetch/Http/ServerSentEvent.php b/src/Fetch/Http/ServerSentEvent.php new file mode 100644 index 0000000..ecfb128 --- /dev/null +++ b/src/Fetch/Http/ServerSentEvent.php @@ -0,0 +1,71 @@ +data`. + * + * @throws \JsonException When the payload is not valid JSON and $throwOnError is true. + */ + public function json(bool $assoc = true, bool $throwOnError = true, int $depth = 512, int $options = 0): mixed + { + try { + return json_decode($this->data, $assoc, $depth, $options | JSON_THROW_ON_ERROR); + } catch (\JsonException $e) { + if ($throwOnError) { + throw $e; + } + + return null; + } + } + + /** + * Determine whether this event's data is the conventional stream + * termination sentinel used by many APIs (e.g. OpenAI's "[DONE]"). + */ + public function isDone(): bool + { + return trim($this->data) === '[DONE]'; + } + + /** + * Get the event data payload when cast to a string. + */ + public function __toString(): string + { + return $this->data; + } +} diff --git a/src/Fetch/Http/StreamedResponse.php b/src/Fetch/Http/StreamedResponse.php new file mode 100644 index 0000000..0d705ed --- /dev/null +++ b/src/Fetch/Http/StreamedResponse.php @@ -0,0 +1,210 @@ +stream()` helpers). The underlying PSR-7 body is + * left unread so callers can pull chunks, lines, or Server-Sent Events as the + * server produces them — the PHP equivalent of JavaScript's `response.body`. + * + * The read methods ({@see stream()}, {@see lines()}) consume the underlying + * stream, so it can only be traversed once. Use {@see buffer()} to fall back + * to a fully materialised {@see Response} when random access is needed. + */ +class StreamedResponse extends BaseResponse implements StreamedResponseInterface +{ + /** + * Create a new streamed response instance. + */ + public static function createFromBase(PsrResponseInterface $response): self + { + return new self( + $response->getStatusCode(), + $response->getHeaders(), + $response->getBody(), + $response->getProtocolVersion(), + $response->getReasonPhrase() + ); + } + + /** + * Yield raw body chunks as they arrive from the underlying stream. + * + * @return Generator + */ + public function stream(int $chunkSize = 8192): Generator + { + if ($chunkSize < 1) { + throw new InvalidArgumentException('Chunk size must be a positive integer.'); + } + + $body = $this->getBody(); + + // Rewind when possible so a fresh iteration starts from the top. + if ($body->isSeekable() && $body->tell() !== 0) { + $body->rewind(); + } + + while (! $body->eof()) { + $chunk = $body->read($chunkSize); + + // A seekable, in-memory stream can legitimately return an empty + // string at EOF without eof() flipping first; guard against a + // busy loop by breaking on empty reads. + if ($chunk === '') { + break; + } + + yield $chunk; + } + } + + /** + * Yield the body one line at a time, handling chunk boundaries. + * + * @return Generator + */ + public function lines(): Generator + { + $buffer = ''; + + foreach ($this->stream() as $chunk) { + $buffer .= $chunk; + + while (($pos = strpos($buffer, "\n")) !== false) { + $line = substr($buffer, 0, $pos); + $buffer = substr($buffer, $pos + 1); + + yield rtrim($line, "\r"); + } + } + + // Emit any trailing content not terminated by a newline. + if ($buffer !== '') { + yield rtrim($buffer, "\r"); + } + } + + /** + * Consume the stream as Server-Sent Events. + */ + public function sse(): EventSource + { + return new EventSource($this); + } + + /** + * Make the streamed response directly iterable over its raw chunks. + * + * @return Generator + */ + public function getIterator(): Generator + { + yield from $this->stream(); + } + + /** + * Drain the remaining stream into a fully buffered {@see Response}. + */ + public function buffer(): Response + { + return new Response( + $this->getStatusCode(), + $this->getHeaders(), + (string) $this->getBody(), + $this->getProtocolVersion(), + $this->getReasonPhrase() + ); + } + + /** + * Get the HTTP status code of the response. + */ + public function status(): int + { + return $this->getStatusCode(); + } + + /** + * Get the status as an enum, or null for unknown codes. + */ + public function statusEnum(): ?Status + { + return Status::tryFrom($this->getStatusCode()); + } + + /** + * Determine whether the response status code is a success (2xx). + */ + public function ok(): bool + { + return $this->getStatusCode() >= 200 && $this->getStatusCode() < 300; + } + + /** + * Determine whether the response is a client or server error (4xx/5xx). + */ + public function failed(): bool + { + return $this->getStatusCode() >= 400; + } + + /** + * Get all response headers. + * + * @return array> + */ + public function headers(): array + { + return $this->getHeaders(); + } + + /** + * Get a single response header line, or null when absent. + */ + public function header(string $header): ?string + { + return $this->hasHeader($header) ? $this->getHeaderLine($header) : null; + } + + /** + * Get the Content-Type header without parameters (e.g. charset). + */ + public function contentType(): ?string + { + $header = $this->getHeaderLine('Content-Type') ?: null; + + if ($header === null) { + return null; + } + + if (($pos = strpos($header, ';')) !== false) { + return trim(substr($header, 0, $pos)); + } + + return $header; + } + + /** + * Get the Content-Type as an enum. + */ + public function contentTypeEnum(): ?ContentType + { + $contentType = $this->contentType(); + + return $contentType !== null ? ContentType::tryFromString($contentType) : null; + } +} diff --git a/src/Fetch/Interfaces/RequestExecutor.php b/src/Fetch/Interfaces/RequestExecutor.php index b900a30..b17b33d 100644 --- a/src/Fetch/Interfaces/RequestExecutor.php +++ b/src/Fetch/Interfaces/RequestExecutor.php @@ -6,6 +6,7 @@ use Fetch\Enum\ContentType; use Fetch\Enum\Method; +use Fetch\Http\EventSource; use React\Promise\PromiseInterface; interface RequestExecutor @@ -67,4 +68,26 @@ public function delete(string $uri, mixed $body = null, string|ContentType $cont * @return Response|PromiseInterface */ public function options(string $uri): Response|PromiseInterface; + + /** + * Send a request and return an unbuffered, streamable response. + * + * @param array $options + */ + public function stream( + Method|string $method = Method::GET, + string $uri = '', + array $options = [], + ): StreamedResponse; + + /** + * Send a request and consume the response as Server-Sent Events. + * + * @param array $options + */ + public function sse( + Method|string $method = Method::GET, + string $uri = '', + array $options = [], + ): EventSource; } diff --git a/src/Fetch/Interfaces/StreamedResponse.php b/src/Fetch/Interfaces/StreamedResponse.php new file mode 100644 index 0000000..75d19d4 --- /dev/null +++ b/src/Fetch/Interfaces/StreamedResponse.php @@ -0,0 +1,102 @@ + + */ +interface StreamedResponse extends \IteratorAggregate, PsrResponseInterface +{ + /** + * Wrap a PSR-7 response for incremental consumption. + */ + public static function createFromBase(PsrResponseInterface $response): self; + + /** + * Yield raw body chunks as they arrive from the underlying stream. + * + * @param int $chunkSize Maximum number of bytes to read per iteration. + * @return Generator + */ + public function stream(int $chunkSize = 8192): Generator; + + /** + * Yield the body one line at a time, handling chunk boundaries. + * + * Line terminators are stripped. Both "\n" and "\r\n" are recognised. + * + * @return Generator + */ + public function lines(): Generator; + + /** + * Consume the stream as Server-Sent Events. + */ + public function sse(): EventSource; + + /** + * Drain the remaining stream into a fully buffered {@see Response}. + * + * This is the escape hatch back to the buffered API. Once buffered, the + * underlying stream is exhausted and cannot be re-read. + */ + public function buffer(): Response; + + /** + * Get the HTTP status code of the response. + */ + public function status(): int; + + /** + * Get the status as an enum, or null for unknown codes. + */ + public function statusEnum(): ?Status; + + /** + * Determine whether the response status code is a success (2xx). + */ + public function ok(): bool; + + /** + * Determine whether the response is a client or server error. + */ + public function failed(): bool; + + /** + * Get all response headers. + * + * @return array> + */ + public function headers(): array; + + /** + * Get a single response header line, or null when absent. + */ + public function header(string $header): ?string; + + /** + * Get the Content-Type header without parameters (e.g. charset). + */ + public function contentType(): ?string; + + /** + * Get the Content-Type as an enum. + */ + public function contentTypeEnum(): ?ContentType; +} diff --git a/src/Fetch/Support/helpers.php b/src/Fetch/Support/helpers.php index 761b6ba..00c93b9 100644 --- a/src/Fetch/Support/helpers.php +++ b/src/Fetch/Support/helpers.php @@ -5,9 +5,11 @@ use Fetch\Enum\ContentType; use Fetch\Enum\Method; use Fetch\Http\Client; +use Fetch\Http\EventSource; use Fetch\Http\Response as HttpResponse; use Fetch\Interfaces\ClientHandler as ClientHandlerInterface; use Fetch\Interfaces\Response as ResponseInterface; +use Fetch\Interfaces\StreamedResponse as StreamedResponseInterface; use Fetch\Support\RequestOptions; use Psr\Http\Client\ClientExceptionInterface; use Psr\Http\Message\RequestInterface; @@ -359,6 +361,64 @@ function delete(string $url, mixed $data = null, ?array $options = []): Response } } +if (! function_exists('fetch_stream')) { + /** + * Perform a request and return an unbuffered, streamable response. + * + * Unlike {@see fetch()}, the response body is not read into memory. Pull + * chunks or lines from the returned {@see StreamedResponse} as the server + * produces them — the PHP equivalent of JavaScript's `response.body`. + * + * ```php + * foreach (fetch_stream('https://example.com/large-file')->lines() as $line) { + * // process each line as it arrives + * } + * ``` + * + * @param string $url URL to fetch + * @param array|null $options Additional request options (method defaults to GET) + */ + function fetch_stream(string $url, ?array $options = []): StreamedResponseInterface + { + $options = $options ?? []; + $method = $options['method'] ?? Method::GET; + unset($options['method']); + + return fetch_client()->stream($url, $options, $method); + } +} + +if (! function_exists('fetch_sse')) { + /** + * Perform a request and consume the response as Server-Sent Events. + * + * Returns an {@see EventSource} that lazily yields {@see ServerSentEvent} + * objects, matching the transport used by streaming LLM APIs and live + * feeds. An `Accept: text/event-stream` header is added automatically. + * + * ```php + * foreach (fetch_sse('https://api.example.com/v1/stream') as $event) { + * if ($event->isDone()) { + * break; + * } + * + * $chunk = $event->json(); + * } + * ``` + * + * @param string $url URL to fetch + * @param array|null $options Additional request options (method defaults to GET) + */ + function fetch_sse(string $url, ?array $options = []): EventSource + { + $options = $options ?? []; + $method = $options['method'] ?? Method::GET; + unset($options['method']); + + return fetch_client()->sse($url, $options, $method); + } +} + // Re-export Matrix async utilities for convenience if (! function_exists('async') && function_exists('\\Matrix\\async')) { /** diff --git a/tests/Integration/StreamingTest.php b/tests/Integration/StreamingTest.php new file mode 100644 index 0000000..01b7b15 --- /dev/null +++ b/tests/Integration/StreamingTest.php @@ -0,0 +1,96 @@ + $stack]); + + return (new ClientHandler)->setHttpClient($guzzle); + } + + public function test_stream_returns_streamed_response(): void + { + $handler = $this->handlerReturning( + new PsrResponse(200, ['Content-Type' => 'text/plain'], Utils::streamFor("chunk-a\nchunk-b\n")) + ); + + $response = $handler->stream(Method::GET, 'https://example.com/feed'); + + $this->assertInstanceOf(StreamedResponse::class, $response); + $this->assertTrue($response->ok()); + $this->assertSame(['chunk-a', 'chunk-b'], iterator_to_array($response->lines(), false)); + } + + public function test_sse_consumes_event_stream_end_to_end(): void + { + $body = "data: {\"n\":1}\n\ndata: {\"n\":2}\n\ndata: [DONE]\n\n"; + + $handler = $this->handlerReturning( + new PsrResponse(200, ['Content-Type' => 'text/event-stream'], Utils::streamFor($body)) + ); + + $source = $handler->sse(Method::GET, 'https://api.example.com/v1/stream'); + + $this->assertInstanceOf(EventSource::class, $source); + + $collected = []; + foreach ($source as $event) { + if ($event->isDone()) { + break; + } + + $collected[] = $event->json(); + } + + $this->assertSame([['n' => 1], ['n' => 2]], $collected); + } + + public function test_sse_adds_accept_header(): void + { + $mock = new MockHandler([ + new PsrResponse(200, ['Content-Type' => 'text/event-stream'], Utils::streamFor("data: ok\n\n")), + ]); + $stack = HandlerStack::create($mock); + $guzzle = new GuzzleClient(['handler' => $stack]); + $handler = (new ClientHandler)->setHttpClient($guzzle); + + iterator_to_array($handler->sse(Method::GET, 'https://api.example.com/stream')->events(), false); + + $this->assertSame('text/event-stream', $mock->getLastRequest()->getHeaderLine('Accept')); + } + + public function test_stream_does_not_buffer_body_until_read(): void + { + $handler = $this->handlerReturning( + new PsrResponse(200, [], Utils::streamFor('lazy-body')) + ); + + $response = $handler->stream(Method::GET, 'https://example.com/x'); + + // Body is still readable on demand (not consumed at construction time). + $this->assertFalse($response->getBody()->eof()); + $this->assertSame('lazy-body', implode('', iterator_to_array($response->stream(), false))); + } +} diff --git a/tests/Unit/ContentTypeTest.php b/tests/Unit/ContentTypeTest.php index bb67870..2a4a235 100644 --- a/tests/Unit/ContentTypeTest.php +++ b/tests/Unit/ContentTypeTest.php @@ -53,9 +53,25 @@ public function test_is_text(): void { $this->assertTrue(ContentType::JSON->isText()); $this->assertTrue(ContentType::TEXT->isText()); + $this->assertTrue(ContentType::EVENT_STREAM->isText()); $this->assertFalse(ContentType::MULTIPART->isText()); } + public function test_is_event_stream(): void + { + $this->assertTrue(ContentType::EVENT_STREAM->isEventStream()); + $this->assertFalse(ContentType::JSON->isEventStream()); + $this->assertSame(ContentType::EVENT_STREAM, ContentType::fromString('text/event-stream')); + } + + public function test_is_streamable(): void + { + $this->assertTrue(ContentType::EVENT_STREAM->isStreamable()); + $this->assertTrue(ContentType::NDJSON->isStreamable()); + $this->assertFalse(ContentType::JSON->isStreamable()); + $this->assertFalse(ContentType::BINARY->isStreamable()); + } + public function test_normalize_content_type(): void { // Test with ContentType instance diff --git a/tests/Unit/EventSourceTest.php b/tests/Unit/EventSourceTest.php new file mode 100644 index 0000000..0c29820 --- /dev/null +++ b/tests/Unit/EventSourceTest.php @@ -0,0 +1,159 @@ + + */ + private function parse(string $body): array + { + $response = new StreamedResponse(200, ['Content-Type' => 'text/event-stream'], Utils::streamFor($body)); + + return iterator_to_array((new EventSource($response))->events(), false); + } + + public function test_parses_simple_events(): void + { + $events = $this->parse("data: hello\n\ndata: world\n\n"); + + $this->assertCount(2, $events); + $this->assertSame('hello', $events[0]->data); + $this->assertSame('world', $events[1]->data); + $this->assertSame('message', $events[0]->type); + } + + public function test_parses_event_type_and_id(): void + { + $events = $this->parse("event: update\nid: 7\ndata: payload\n\n"); + + $this->assertCount(1, $events); + $this->assertSame('update', $events[0]->type); + $this->assertSame('7', $events[0]->id); + $this->assertSame('payload', $events[0]->data); + } + + public function test_joins_multiline_data(): void + { + $events = $this->parse("data: line one\ndata: line two\n\n"); + + $this->assertCount(1, $events); + $this->assertSame("line one\nline two", $events[0]->data); + } + + public function test_ignores_comment_lines(): void + { + $events = $this->parse(": this is a comment\ndata: real\n\n"); + + $this->assertCount(1, $events); + $this->assertSame('real', $events[0]->data); + } + + public function test_strips_only_first_leading_space_from_value(): void + { + $events = $this->parse("data: two spaces\n\n"); + + $this->assertSame(' two spaces', $events[0]->data); + } + + public function test_value_without_space_after_colon(): void + { + $events = $this->parse("data:nospace\n\n"); + + $this->assertSame('nospace', $events[0]->data); + } + + public function test_field_without_colon_is_empty_value(): void + { + // A lone "data" line contributes an empty data line. + $events = $this->parse("data\n\n"); + + $this->assertCount(1, $events); + $this->assertSame('', $events[0]->data); + } + + public function test_events_without_data_do_not_dispatch(): void + { + $events = $this->parse("event: ping\nid: 1\n\ndata: kept\n\n"); + + $this->assertCount(1, $events); + $this->assertSame('kept', $events[0]->data); + } + + public function test_last_event_id_persists_across_events(): void + { + $events = $this->parse("id: 100\ndata: a\n\ndata: b\n\n"); + + $this->assertSame('100', $events[0]->id); + $this->assertSame('100', $events[1]->id); + } + + public function test_retry_field_parsed_and_carried(): void + { + $events = $this->parse("retry: 5000\ndata: a\n\n"); + + $this->assertSame(5000, $events[0]->retry); + } + + public function test_non_numeric_retry_is_ignored(): void + { + $events = $this->parse("retry: soon\ndata: a\n\n"); + + $this->assertNull($events[0]->retry); + } + + public function test_incomplete_trailing_event_is_discarded(): void + { + // No terminating blank line, so the final event is not dispatched. + $events = $this->parse("data: complete\n\ndata: dangling\n"); + + $this->assertCount(1, $events); + $this->assertSame('complete', $events[0]->data); + } + + public function test_done_sentinel_is_yielded_as_event(): void + { + $events = $this->parse("data: {\"x\":1}\n\ndata: [DONE]\n\n"); + + $this->assertCount(2, $events); + $this->assertFalse($events[0]->isDone()); + $this->assertTrue($events[1]->isDone()); + } + + public function test_events_split_across_stream_chunks(): void + { + // Build a body that will straddle the 8192-byte default chunk size, + // forcing the line buffer to reassemble events across reads. + $body = ''; + for ($i = 0; $i < 2000; $i++) { + $body .= "data: event-{$i}\n\n"; + } + + $events = $this->parse($body); + + $this->assertCount(2000, $events); + $this->assertSame('event-0', $events[0]->data); + $this->assertSame('event-1999', $events[1999]->data); + } + + public function test_tracks_last_event_id_and_reconnection_time(): void + { + $response = new StreamedResponse(200, [], Utils::streamFor("retry: 2500\nid: abc\ndata: x\n\n")); + $source = new EventSource($response); + + iterator_to_array($source->events(), false); + + $this->assertSame('abc', $source->lastEventId()); + $this->assertSame(2500, $source->reconnectionTime()); + $this->assertSame($response, $source->response()); + } +} diff --git a/tests/Unit/ServerSentEventTest.php b/tests/Unit/ServerSentEventTest.php new file mode 100644 index 0000000..a44f7cc --- /dev/null +++ b/tests/Unit/ServerSentEventTest.php @@ -0,0 +1,65 @@ +assertSame('', $event->data); + $this->assertSame('message', $event->type); + $this->assertNull($event->id); + $this->assertNull($event->retry); + } + + public function test_exposes_all_fields(): void + { + $event = new ServerSentEvent(data: 'hello', type: 'greeting', id: '42', retry: 3000); + + $this->assertSame('hello', $event->data); + $this->assertSame('greeting', $event->type); + $this->assertSame('42', $event->id); + $this->assertSame(3000, $event->retry); + } + + public function test_json_decodes_payload(): void + { + $event = new ServerSentEvent(data: '{"delta":"hi","index":0}'); + + $this->assertSame(['delta' => 'hi', 'index' => 0], $event->json()); + $this->assertSame('hi', $event->json()->delta ?? ($event->json(false)->delta)); + } + + public function test_json_returns_null_on_invalid_when_not_throwing(): void + { + $event = new ServerSentEvent(data: 'not-json'); + + $this->assertNull($event->json(throwOnError: false)); + } + + public function test_json_throws_on_invalid_by_default(): void + { + $this->expectException(\JsonException::class); + + (new ServerSentEvent(data: 'not-json'))->json(); + } + + public function test_is_done_detects_sentinel(): void + { + $this->assertTrue((new ServerSentEvent(data: '[DONE]'))->isDone()); + $this->assertTrue((new ServerSentEvent(data: ' [DONE] '))->isDone()); + $this->assertFalse((new ServerSentEvent(data: 'done'))->isDone()); + } + + public function test_stringable(): void + { + $this->assertSame('payload', (string) new ServerSentEvent(data: 'payload')); + } +} diff --git a/tests/Unit/StreamedResponseTest.php b/tests/Unit/StreamedResponseTest.php new file mode 100644 index 0000000..08cce66 --- /dev/null +++ b/tests/Unit/StreamedResponseTest.php @@ -0,0 +1,108 @@ +make('abcdefghij'); + + $chunks = iterator_to_array($response->stream(4), false); + + $this->assertSame(['abcd', 'efgh', 'ij'], $chunks); + } + + public function test_stream_rejects_non_positive_chunk_size(): void + { + $this->expectException(\InvalidArgumentException::class); + + iterator_to_array($this->make('x')->stream(0)); + } + + public function test_yields_lines_across_chunk_boundaries(): void + { + $response = $this->make("first line\r\nsecond line\nthird"); + + $lines = iterator_to_array($response->lines(), false); + + $this->assertSame(['first line', 'second line', 'third'], $lines); + } + + public function test_lines_handles_trailing_newline(): void + { + $response = $this->make("only\n"); + + $this->assertSame(['only'], iterator_to_array($response->lines(), false)); + } + + public function test_buffer_returns_full_response(): void + { + $buffered = $this->make('{"ok":true}', ['Content-Type' => 'application/json'])->buffer(); + + $this->assertInstanceOf(Response::class, $buffered); + $this->assertSame(['ok' => true], $buffered->json()); + } + + public function test_status_helpers(): void + { + $ok = $this->make('', [], 200); + $this->assertTrue($ok->ok()); + $this->assertFalse($ok->failed()); + $this->assertSame(Status::OK, $ok->statusEnum()); + + $error = $this->make('', [], 503); + $this->assertFalse($error->ok()); + $this->assertTrue($error->failed()); + $this->assertSame(503, $error->status()); + } + + public function test_header_accessors(): void + { + $response = $this->make('', ['Content-Type' => 'text/event-stream; charset=utf-8']); + + $this->assertSame('text/event-stream; charset=utf-8', $response->header('Content-Type')); + $this->assertNull($response->header('X-Missing')); + $this->assertSame('text/event-stream', $response->contentType()); + $this->assertSame(ContentType::EVENT_STREAM, $response->contentTypeEnum()); + } + + public function test_is_iterable_over_chunks(): void + { + $response = $this->make('hello world'); + + $this->assertSame('hello world', implode('', iterator_to_array($response, false))); + } + + public function test_sse_returns_event_source(): void + { + $this->assertInstanceOf(EventSource::class, $this->make('')->sse()); + } + + public function test_create_from_base_preserves_metadata(): void + { + $base = new PsrResponse(201, ['X-Test' => 'yes'], 'body', '2', 'Created'); + + $response = StreamedResponse::createFromBase($base); + + $this->assertSame(201, $response->status()); + $this->assertSame('yes', $response->header('X-Test')); + $this->assertSame('2', $response->getProtocolVersion()); + } +}