File Server/Transport/StreamableHttpServerTransport.php

1:	<?php
2:
3:	declare(strict_types=1);
4:
5:	/**
6:	* This file is part of the Nexus MCP SDK package.
7:	*
8:	* (c) 2026 John Paul E. Balandan, CPA <paulbalandan@gmail.com>
9:	*
10:	* For the full copyright and license information, please view
11:	* the LICENSE file that was distributed with this source code.
12:	*/
13:
14:	namespace Nexus\Mcp\Server\Transport;
15:
16:	use Amp\DeferredFuture;
17:	use Nexus\Assert\Assert;
18:	use Nexus\Mcp\Core\Auth\VerifiedAccessToken;
19:	use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
20:	use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
21:	use Nexus\Mcp\Core\Exception\TransportNotStartedException;
22:	use Nexus\Mcp\Core\Http\HttpStatus;
23:	use Nexus\Mcp\Core\Http\HttpStatusResolver;
24:	use Nexus\Mcp\Core\Http\StandardHeaders;
25:	use Nexus\Mcp\Core\JsonRpc\EnvelopeRequestId;
26:	use Nexus\Mcp\Core\Schema\Error;
27:	use Nexus\Mcp\Core\Schema\Error\InternalError;
28:	use Nexus\Mcp\Core\Schema\Error\InvalidRequestError;
29:	use Nexus\Mcp\Core\Schema\Error\ParseError;
30:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcErrorResponse;
31:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
32:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcNotification;
33:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcResultResponse;
34:	use Nexus\Mcp\Core\Schema\Notification\CancelledNotification;
35:	use Nexus\Mcp\Core\Schema\Request\SubscriptionsListenRequest;
36:	use Nexus\Mcp\Core\Schema\RequestId;
37:	use Nexus\Mcp\Core\Transport\CancellableTransportInterface;
38:	use Nexus\Mcp\Core\Transport\ListenerHandle;
39:	use Nexus\Mcp\Core\Transport\ListenerHandleInterface;
40:	use Nexus\Mcp\Core\Transport\ReceiveContext;
41:	use Nexus\Mcp\Core\Transport\SendContext;
42:	use Nexus\Mcp\Core\Transport\TransportEvents;
43:	use Nexus\Mcp\Core\Transport\TransportState;
44:	use Nexus\Mcp\Server\Transport\Http\ResponseMode;
45:	use Nexus\Mcp\Server\Transport\Http\SseResponseStream;
46:	use Psr\Http\Message\ResponseFactoryInterface;
47:	use Psr\Http\Message\ResponseInterface;
48:	use Psr\Http\Message\ServerRequestInterface;
49:	use Psr\Http\Message\StreamFactoryInterface;
50:	use Psr\Http\Server\RequestHandlerInterface;
51:	use Psr\Log\LoggerInterface;
52:	use Psr\Log\NullLogger;
53:
54:	/**
55:	* Stateless Streamable HTTP server transport, and the PSR-15 request handler for the MCP endpoint.
56:	*
57:	* @see https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http
58:	*/
59:	final class StreamableHttpServerTransport implements CancellableTransportInterface, RequestHandlerInterface
60:	{
61:	/**
62:	* PSR-7 request attribute a middleware leaves the decoded JSON-RPC envelope on.
63:	*/
64:	public const string ENVELOPE_ATTRIBUTE = 'nexus.mcp.envelope';
65:
66:	private const int DEFAULT_MAX_BUFFERED_BYTES = 1_048_576;
67:
68:	private TransportState $state = TransportState::Idle;
69:
70:	/**
71:	* True from the first `close()` on, which `state` cannot signal as it stays `Running` across the
72:	* drain so a listener may still send.
73:	*/
74:	private bool $closing = false;
75:
76:	/**
77:	* @var null\|\Fiber<mixed, mixed, mixed, mixed> The close owner, `null` while no close is in progress or when {main} owns it
78:	*/
79:	private ?\Fiber $closingFiber = null;
80:
81:	/**
82:	* @var null\|DeferredFuture<null>
83:	*/
84:	private ?DeferredFuture $closeCompletion = null;
85:
86:	private readonly TransportEvents $events;
87:	private readonly StandardHeaders $standardHeaders;
88:
89:	/**
90:	* The last transport-internal request id minted, ascending and never reused, so a retired sink's id
91:	* cannot reach a later request.
92:	*/
93:	private int $lastRequestId = 0;
94:
95:	/**
96:	* In-flight requests keyed by the transport-internal id, the `buffered` deferred carrying the response
97:	* `handle()` awaits until `stream` is set and SSE frames take over.
98:	*
99:	* @var array<int, array{
100:	* clientId: int\|non-empty-string,
101:	* buffered: DeferredFuture<ResponseInterface>,
102:	* stream: null\|SseResponseStream,
103:	* }>
104:	*/
105:	private array $sinks = [];
106:
107:	/**
108:	* @var array<int, \Closure(RequestId): void>
109:	*/
110:	private array $cancelListeners = [];
111:
112:	/**
113:	* @param int<1, max> $maxBufferedBytes Unread SSE bytes past which a further frame abandons the stream
114:	*/
115:	public function __construct(
116:	private readonly ResponseFactoryInterface $responseFactory,
117:	private readonly StreamFactoryInterface $streamFactory,
118:	private readonly LoggerInterface $logger = new NullLogger(),
119:	private readonly ResponseMode $responseMode = ResponseMode::Auto,
120:	private readonly float $keepAliveInterval = 15.0,
121:	private readonly int $maxBufferedBytes = self::DEFAULT_MAX_BUFFERED_BYTES,
122:	) {
123:	if ($this->keepAliveInterval <= 0.0) {
124:	throw new \InvalidArgumentException(\sprintf('The SSE keep-alive interval must be positive, %s given.', $this->keepAliveInterval));
125:	}
126:
127:	Assert::that($this->maxBufferedBytes)->isPositiveInt('The SSE buffer cap must be positive, {value} given.');
128:
129:	$this->events = TransportEvents::create($this->logger, 'Streamable HTTP server');
130:	$this->standardHeaders = new StandardHeaders();
131:	}
132:
133:	#[\Override]
134:	public function handle(ServerRequestInterface $request): ResponseInterface
135:	{
136:	if ($request->getMethod() !== 'POST') {
137:	return $this->responseFactory->createResponse(HttpStatus::MethodNotAllowed->value)->withHeader('Allow', 'POST');
138:	}
139:
140:	if (TransportState::Running !== $this->state) {
141:	return $this->buildErrorResponse(
142:	new InternalError(message: 'The MCP endpoint is not accepting requests.'),
143:	HttpStatus::ServiceUnavailable->value,
144:	);
145:	}
146:
147:	if (! $this->acceptsRequiredContentTypes($request)) {
148:	return $this->responseFactory->createResponse(HttpStatus::NotAcceptable->value);
149:	}
150:
151:	$envelope = $request->getAttribute(self::ENVELOPE_ATTRIBUTE);
152:
153:	if (! \is_array($envelope)) {
154:	try {
155:	$envelope = json_decode((string) $request->getBody(), associative: true, flags: \JSON_THROW_ON_ERROR);
156:	} catch (\JsonException $e) {
157:	$this->logger->warning('Streamable HTTP server transport rejected a malformed JSON body.', ['exception' => $e]);
158:	$this->events->emitError($e);
159:
160:	return $this->buildErrorResponse(new ParseError(message: ParseError::DEFAULT_MESSAGE));
161:	}
162:	}
163:
164:	try {
165:	Assert::that($envelope)->isMap('JSON-RPC envelope must be a JSON object, {type} given.');
166:	} catch (\InvalidArgumentException $e) {
167:	$this->logger->warning('Streamable HTTP server transport rejected a non-object envelope.', ['exception' => $e]);
168:	$this->events->emitError($e);
169:
170:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
171:	}
172:
173:	if (! \array_key_exists('method', $envelope) && (\array_key_exists('result', $envelope) \|\| \array_key_exists('error', $envelope))) {
174:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
175:	}
176:
177:	if (! \array_key_exists('id', $envelope)) {
178:	if (! $this->isAcceptableNotification($envelope)) {
179:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
180:	}
181:
182:	if (CancelledNotification::getMethod() === ($envelope['method'] ?? null)) {
183:	$this->logger->debug(
184:	'Streamable HTTP server transport ignored a client cancellation notification: the response stream is the signal on this transport.',
185:	);
186:
187:	return $this->responseFactory->createResponse(HttpStatus::Accepted->value);
188:	}
189:
190:	$this->events->emitMessage($envelope, $this->buildReceiveContext($request));
191:
192:	return $this->responseFactory->createResponse(HttpStatus::Accepted->value);
193:	}
194:
195:	$requestId = EnvelopeRequestId::recover($envelope);
196:
197:	if (null === $requestId) {
198:	return $this->buildErrorResponse(new InvalidRequestError(message: InvalidRequestError::DEFAULT_MESSAGE));
199:	}
200:
201:	$clientId = $requestId->id;
202:
203:	$mismatch = $this->standardHeaders->validate($this->readHeaders($request), $envelope);
204:
205:	if (null !== $mismatch) {
206:	return $this->buildErrorResponse($mismatch, id: $requestId);
207:	}
208:
209:	$streams = ResponseMode::Sse === $this->responseMode
210:	\|\| SubscriptionsListenRequest::getMethod() === ($envelope['method'] ?? null);
211:
212:	return $streams
213:	? $this->dispatchStreaming($envelope, $clientId, $request)
214:	: $this->dispatchBuffered($envelope, $clientId, $request);
215:	}
216:
217:	#[\Override]
218:	public function start(): void
219:	{
220:	match ($this->state) {
221:	TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
222:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
223:	TransportState::Idle => null,
224:	};
225:
226:	$this->state = TransportState::Running;
227:	$this->logger->info('Streamable HTTP server transport started.');
228:	}
229:
230:	#[\Override]
231:	public function send(JsonRpcMessage $message, ?SendContext $context = null): void
232:	{
233:	match ($this->state) {
234:	TransportState::Idle => throw new TransportNotStartedException(operation: 'send'),
235:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'send'),
236:	TransportState::Running => null,
237:	};
238:
239:	if ($message instanceof JsonRpcResultResponse \|\| $message instanceof JsonRpcErrorResponse) {
240:	$this->routeResponse($message, $context);
241:
242:	return;
243:	}
244:
245:	if ($message instanceof JsonRpcNotification) {
246:	$this->routeNotification($message, $context);
247:
248:	return;
249:	}
250:
251:	$this->logger->warning('Streamable HTTP server transport dropped an unexpected server-initiated request.');
252:	}
253:
254:	#[\Override]
255:	public function close(): void
256:	{
257:	if ($this->closing) {
258:	if (\Fiber::getCurrent() === $this->closingFiber) {
259:	return;
260:	}
261:
262:	$this->closeCompletion?->getFuture()->await();
263:
264:	return;
265:	}
266:
267:	$this->closing = true;
268:	$this->closingFiber = \Fiber::getCurrent();
269:
270:	/** @var DeferredFuture<null> $completion */
271:	$completion = new DeferredFuture();
272:	$this->closeCompletion = $completion;
273:
274:	try {
275:	$this->events->emitDrain();
276:	} finally {
277:	$this->state = TransportState::Closed;
278:
279:	try {
280:	try {
281:	$this->retireSinks();
282:	} finally {
283:	$this->events->emitClose();
284:	$this->logger->info('Streamable HTTP server transport closed.');
285:	}
286:	} finally {
287:	$completion->complete();
288:	}
289:	}
290:	}
291:
292:	#[\Override]
293:	public function onMessage(\Closure $listener): ListenerHandleInterface
294:	{
295:	return $this->events->onMessage($listener);
296:	}
297:
298:	#[\Override]
299:	public function onError(\Closure $listener): ListenerHandleInterface
300:	{
301:	return $this->events->onError($listener);
302:	}
303:
304:	#[\Override]
305:	public function onDrain(\Closure $listener): ListenerHandleInterface
306:	{
307:	return $this->events->onDrain($listener);
308:	}
309:
310:	#[\Override]
311:	public function onCancel(\Closure $listener): ListenerHandleInterface
312:	{
313:	$id = spl_object_id($listener);
314:	$this->cancelListeners[$id] = $listener;
315:
316:	return new ListenerHandle(function () use ($id): void {
317:	unset($this->cancelListeners[$id]);
318:	});
319:	}
320:
321:	#[\Override]
322:	public function onClose(\Closure $listener): ListenerHandleInterface
323:	{
324:	return $this->events->onClose($listener);
325:	}
326:
327:	/**
328:	* The client MUST accept both media types (`Accept: application/json, text/event-stream`) so the server
329:	* is free to answer with a buffered JSON object or an SSE stream.
330:	*/
331:	private function acceptsRequiredContentTypes(ServerRequestInterface $request): bool
332:	{
333:	$ranges = $this->parseAcceptableMediaRanges($request->getHeaderLine('Accept'));
334:
335:	return $this->matchesMediaRange($ranges, 'application/json') && $this->matchesMediaRange($ranges, 'text/event-stream');
336:	}
337:
338:	/**
339:	* The RFC 9110 media ranges an `Accept` header lists with a positive quality.
340:	*
341:	* @return list<string>
342:	*/
343:	private function parseAcceptableMediaRanges(string $accept): array
344:	{
345:	$ranges = [];
346:
347:	foreach (explode(',', $accept) as $element) {
348:	$parts = explode(';', $element);
349:
350:	foreach ($parts as $parameter) {
351:	if (preg_match('/\A\sq\s=\s0(?:\.0{0,3})?\s\z/i', $parameter) === 1) {
352:	continue 2;
353:	}
354:	}
355:
356:	$ranges[] = strtolower(trim($parts[0]));
357:	}
358:
359:	return $ranges;
360:	}
361:
362:	/**
363:	* @param list<string> $ranges
364:	* @param non-empty-string $mediaType
365:	*/
366:	private function matchesMediaRange(array $ranges, string $mediaType): bool
367:	{
368:	$typeWildcard = strstr($mediaType, '/', true).'/*';
369:
370:	foreach ($ranges as $range) {
371:	if ($range === $mediaType \|\| '/' === $range \|\| $range === $typeWildcard) {
372:	return true;
373:	}
374:	}
375:
376:	return false;
377:	}
378:
379:	/**
380:	* A notification receives no dispatcher reply, so the transport gates its acceptance itself: a
381:	* well-formed JSON-RPC 2.0 notification carries a non-empty string method.
382:	*
383:	* @param array<string, mixed> $envelope
384:	*/
385:	private function isAcceptableNotification(array $envelope): bool
386:	{
387:	$method = $envelope['method'] ?? null;
388:
389:	return JsonRpcMessage::JSONRPC_VERSION === ($envelope['jsonrpc'] ?? null)
390:	&& \is_string($method)
391:	&& '' !== $method;
392:	}
393:
394:	/**
395:	* Buffers the request into a JSON response, which under `Auto` the first progress notification
396:	* upgrades to an SSE stream.
397:	*
398:	* @param array<string, mixed> $envelope
399:	* @param int\|non-empty-string $clientId
400:	*/
401:	private function dispatchBuffered(array $envelope, int\|string $clientId, ServerRequestInterface $request): ResponseInterface
402:	{
403:	/** @var DeferredFuture<ResponseInterface> $deferred */
404:	$deferred = new DeferredFuture();
405:
406:	$internalId = ++$this->lastRequestId;
407:	$this->sinks[$internalId] = ['clientId' => $clientId, 'buffered' => $deferred, 'stream' => null];
408:
409:	$this->emitRequest($envelope, $internalId, $this->buildReceiveContext($request, $clientId));
410:
411:	return $deferred->getFuture()->await();
412:	}
413:
414:	/**
415:	* Answers immediately with an SSE stream the dispatch coroutine writes progress frames and the final
416:	* response to.
417:	*
418:	* @param array<string, mixed> $envelope
419:	* @param int\|non-empty-string $clientId
420:	*/
421:	private function dispatchStreaming(array $envelope, int\|string $clientId, ServerRequestInterface $request): ResponseInterface
422:	{
423:	/** @var DeferredFuture<ResponseInterface> $unused */
424:	$unused = new DeferredFuture();
425:	$internalId = ++$this->lastRequestId;
426:	$stream = $this->buildStream($internalId);
427:	$response = $this->buildSseResponse($stream);
428:	$this->sinks[$internalId] = ['clientId' => $clientId, 'buffered' => $unused, 'stream' => $stream];
429:
430:	$this->emitRequest($envelope, $internalId, $this->buildReceiveContext($request, $clientId));
431:
432:	return $response;
433:	}
434:
435:	/**
436:	* Emits an inbound request under its transport-internal id, retiring its sink when a listener throws.
437:	*
438:	* @param array<string, mixed> $envelope
439:	*/
440:	private function emitRequest(array $envelope, int $internalId, ReceiveContext $context): void
441:	{
442:	$envelope['id'] = $internalId;
443:
444:	try {
445:	$this->events->emitMessage($envelope, $context);
446:	} catch (\Throwable $e) {
447:	unset($this->sinks[$internalId]);
448:
449:	throw $e;
450:	}
451:	}
452:
453:	/**
454:	* Retires every request still in flight at close: an open SSE stream reaches end-of-body, and a
455:	* buffered one gets a service-unavailable error. A sink whose error cannot be built fails with the
456:	* cause instead, the last of which surfaces once the rest are retired.
457:	*/
458:	private function retireSinks(): void
459:	{
460:	$sinks = $this->sinks;
461:	$this->sinks = [];
462:	$failure = null;
463:
464:	foreach ($sinks as $sink) {
465:	$stream = $sink['stream'];
466:
467:	if (null !== $stream) {
468:	$stream->end();
469:
470:	continue;
471:	}
472:
473:	try {
474:	$sink['buffered']->complete($this->buildErrorResponse(
475:	new InternalError(message: 'The MCP endpoint is shutting down.'),
476:	status: HttpStatus::ServiceUnavailable->value,
477:	id: new RequestId(id: $sink['clientId']),
478:	));
479:	} catch (\Throwable $e) {
480:	$sink['buffered']->error($e);
481:	$failure = $e;
482:	}
483:	}
484:
485:	if (null !== $failure) {
486:	throw $failure;
487:	}
488:	}
489:
490:	/**
491:	* Carries the HTTP request, and the token a bearer-authentication stage validated it with, to handlers.
492:	*
493:	* @param null\|int\|non-empty-string $clientId
494:	*/
495:	private function buildReceiveContext(ServerRequestInterface $request, null\|int\|string $clientId = null): ReceiveContext
496:	{
497:	$token = $request->getAttribute(VerifiedAccessToken::REQUEST_ATTRIBUTE);
498:
499:	return new ReceiveContext(
500:	$request,
501:	$token instanceof VerifiedAccessToken ? $token : null,
502:	null === $clientId ? null : new RequestId(id: $clientId),
503:	);
504:	}
505:
506:	private function routeResponse(JsonRpcErrorResponse\|JsonRpcResultResponse $message, ?SendContext $context): void
507:	{
508:	$id = $message->id;
509:
510:	if (! $id instanceof RequestId) {
511:	$this->logger->warning('Streamable HTTP server transport discarded a response that carries no id to correlate.');
512:
513:	return;
514:	}
515:
516:	$internalId = $id->id;
517:
518:	if (! \is_int($internalId) \|\| ! \array_key_exists($internalId, $this->sinks)) {
519:	$this->logger->warning('Streamable HTTP server transport discarded an orphan response with no in-flight request.');
520:
521:	return;
522:	}
523:
524:	$sink = $this->sinks[$internalId];
525:	$envelope = $message->jsonSerialize();
526:	$envelope['id'] = $sink['clientId'];
527:	$status = $this->resolveStatus($message, null !== $context && $context->fromHandler);
528:	$failure = null;
529:
530:	try {
531:	$payload = $this->encode($envelope);
532:	} catch (\JsonException $e) {
533:	$payload = $this->encode($this->buildUnencodableError($sink['clientId']));
534:	$status = HttpStatus::InternalServerError->value;
535:	$failure = $e;
536:	}
537:
538:	$this->deliver($sink, $internalId, $payload, $status);
539:
540:	if (null !== $failure) {
541:	$this->logger->error('Streamable HTTP server transport replaced a response JSON cannot encode with an internal error: {reason}.', ['reason' => $failure->getMessage()]);
542:	}
543:	}
544:
545:	/**
546:	* Writes the payload to its request's sink, retiring it first so nothing else can settle the request.
547:	* A response that cannot be built fails the request rather than leaving it parked.
548:	*
549:	* @param array{clientId: int\|non-empty-string, buffered: DeferredFuture<ResponseInterface>, stream: null\|SseResponseStream} $sink
550:	*/
551:	private function deliver(array $sink, int $internalId, string $payload, int $status): void
552:	{
553:	$stream = $sink['stream'];
554:	unset($this->sinks[$internalId]);
555:
556:	if (null !== $stream) {
557:	$stream->push($this->frame($payload));
558:	$stream->end();
559:	} else {
560:	try {
561:	$response = $this->buildJsonResponse($payload, $status);
562:	} catch (\Throwable $e) {
563:	$sink['buffered']->error($e);
564:
565:	throw $e;
566:	}
567:
568:	$sink['buffered']->complete($response);
569:	}
570:	}
571:
572:	/**
573:	* The error standing in for a response JSON cannot encode, echoing the client's own id.
574:	*
575:	* @param int\|non-empty-string $clientId
576:	*
577:	* @return array<string, mixed>
578:	*/
579:	private function buildUnencodableError(int\|string $clientId): array
580:	{
581:	return (new JsonRpcErrorResponse(
582:	id: new RequestId(id: $clientId),
583:	error: new InternalError(message: 'The response could not be encoded.'),
584:	))->jsonSerialize();
585:	}
586:
587:	/**
588:	* @param JsonRpcNotification<non-empty-string> $notification
589:	*/
590:	private function routeNotification(JsonRpcNotification $notification, ?SendContext $context): void
591:	{
592:	$related = $context?->relatedRequestId;
593:
594:	if (! $related instanceof RequestId) {
595:	$this->logger->debug('Streamable HTTP server transport dropped a notification with no related request to stream it to.');
596:
597:	return;
598:	}
599:
600:	$internalId = $related->id;
601:
602:	if (! \is_int($internalId) \|\| ! \array_key_exists($internalId, $this->sinks)) {
603:	$this->logger->debug('Streamable HTTP server transport dropped a notification for a request that is no longer in flight.');
604:
605:	return;
606:	}
607:
608:	$sink = $this->sinks[$internalId];
609:	$stream = $sink['stream'];
610:
611:	if (null !== $stream) {
612:	$stream->push($this->frame($this->encode($notification->jsonSerialize())));
613:	} elseif (ResponseMode::Auto === $this->responseMode) {
614:	$this->upgradeToStream($internalId, $sink['clientId'], $sink['buffered'], $notification->jsonSerialize());
615:	} else {
616:	$this->logger->debug('Streamable HTTP server transport dropped a notification: the JSON response mode cannot stream it.');
617:	}
618:	}
619:
620:	/**
621:	* Lazily promotes a buffered `Auto` request to an SSE stream, standing down when the request was
622:	* settled while its response was being built.
623:	*
624:	* @param int\|non-empty-string $clientId
625:	* @param DeferredFuture<ResponseInterface> $buffered
626:	* @param array<string, mixed> $envelope
627:	*/
628:	private function upgradeToStream(int $internalId, int\|string $clientId, DeferredFuture $buffered, array $envelope): void
629:	{
630:	$frame = $this->frame($this->encode($envelope));
631:	$stream = $this->buildStream($internalId);
632:	$response = $this->buildSseResponse($stream);
633:
634:	if (! \array_key_exists($internalId, $this->sinks)) {
635:	return;
636:	}
637:
638:	$this->sinks[$internalId] = ['clientId' => $clientId, 'buffered' => $buffered, 'stream' => $stream];
639:
640:	$stream->push($frame);
641:	$buffered->complete($response);
642:	}
643:
644:	private function buildStream(int $internalId): SseResponseStream
645:	{
646:	return new SseResponseStream(
647:	$this->keepAliveInterval,
648:	$this->maxBufferedBytes,
649:	function (bool $overflowed) use ($internalId): void {
650:	if ($overflowed) {
651:	$this->logger->warning(
652:	'Streamable HTTP server transport abandoned a stream whose reader fell at least {limit} bytes behind.',
653:	['limit' => $this->maxBufferedBytes],
654:	);
655:	}
656:
657:	$this->releaseStream($internalId);
658:	},
659:	);
660:	}
661:
662:	/**
663:	* Retires a stream whose body the consumer closed (a client disconnect or a buffer overflow), and a no-op
664:	* once it has ended.
665:	*/
666:	private function releaseStream(int $internalId): void
667:	{
668:	$sink = $this->sinks[$internalId] ?? null;
669:	unset($this->sinks[$internalId]);
670:
671:	if (null === $sink) {
672:	return;
673:	}
674:
675:	$abandoned = new RequestId(id: $internalId);
676:
677:	foreach ($this->cancelListeners as $listener) {
678:	$listener($abandoned);
679:	}
680:	}
681:
682:	/**
683:	* A result and a handler-produced error both ride HTTP 200 with the JSON-RPC payload in the body,
684:	* while a protocol error carries a real status.
685:	*/
686:	private function resolveStatus(JsonRpcErrorResponse\|JsonRpcResultResponse $message, bool $fromHandler): int
687:	{
688:	if ($message instanceof JsonRpcResultResponse) {
689:	return HttpStatus::Ok->value;
690:	}
691:
692:	return HttpStatusResolver::resolve($message->error->code, $fromHandler);
693:	}
694:
695:	private function buildJsonResponse(string $payload, int $status): ResponseInterface
696:	{
697:	return $this->responseFactory->createResponse($status)
698:	->withHeader('Content-Type', 'application/json')
699:	->withBody($this->streamFactory->createStream($payload))
700:	;
701:	}
702:
703:	private function buildSseResponse(SseResponseStream $body): ResponseInterface
704:	{
705:	return $this->responseFactory->createResponse(HttpStatus::Ok->value)
706:	->withHeader('Content-Type', 'text/event-stream')
707:	->withHeader('Cache-Control', 'no-cache')
708:	->withHeader('Connection', 'keep-alive')
709:	->withHeader('X-Accel-Buffering', 'no')
710:	->withBody($body)
711:	;
712:	}
713:
714:	/**
715:	* @param null\|int $status The HTTP status to pin, or `null` to derive it from the error's code
716:	*/
717:	private function buildErrorResponse(Error $error, ?int $status = null, ?RequestId $id = null): ResponseInterface
718:	{
719:	$status ??= HttpStatusResolver::resolve($error->code, fromHandler: false);
720:	$envelope = (new JsonRpcErrorResponse(id: $id, error: $error))->jsonSerialize();
721:
722:	return $this->responseFactory->createResponse($status)
723:	->withHeader('Content-Type', 'application/json')
724:	->withBody($this->streamFactory->createStream($this->encode($envelope)))
725:	;
726:	}
727:
728:	/**
729:	* @return array<string, string>
730:	*/
731:	private function readHeaders(ServerRequestInterface $request): array
732:	{
733:	return array_map(
734:	static fn(array $values): string => implode(', ', $values),
735:	$request->getHeaders(),
736:	);
737:	}
738:
739:	private function frame(string $payload): string
740:	{
741:	return \sprintf("event: message\ndata: %s\n\n", $payload);
742:	}
743:
744:	/**
745:	* @param array<string, mixed> $envelope
746:	*/
747:	private function encode(array $envelope): string
748:	{
749:	return json_encode($envelope, \JSON_THROW_ON_ERROR \| \JSON_UNESCAPED_SLASHES \| \JSON_UNESCAPED_UNICODE);
750:	}
751:	}
752:

Namespaces

Classes