File Core/Transport/InMemoryTransport.php | Nexus MCP SDK (1.x-dev) API

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\Core\Transport;
15:
16:	use Amp\DeferredFuture;
17:	use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
18:	use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
19:	use Nexus\Mcp\Core\Exception\TransportNotStartedException;
20:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
21:
22:	/**
23:	* In-process JSON-RPC duplex between two `TransportInterface` instances.
24:	*/
25:	final class InMemoryTransport implements TransportInterface
26:	{
27:	/**
28:	* Envelopes the peer's `send()` delivered before this side called `start()`, drained there in arrival order.
29:	*
30:	* @var list<array<string, mixed>>
31:	*/
32:	private array $pendingInbound = [];
33:
34:	private TransportState $state = TransportState::Idle;
35:
36:	/**
37:	* True from the first `close()` on, which `state` cannot signal as it stays `Running` across the
38:	* drain so a listener may still send.
39:	*/
40:	private bool $closing = false;
41:
42:	/**
43:	* @var null\|\Fiber<mixed, mixed, mixed, mixed> The close owner, `null` while no close is in progress or when {main} owns it
44:	*/
45:	private ?\Fiber $closingFiber = null;
46:
47:	/**
48:	* @var null\|DeferredFuture<null>
49:	*/
50:	private ?DeferredFuture $closeCompletion = null;
51:
52:	private ?self $peer = null;
53:	private readonly TransportEvents $events;
54:
55:	private function __construct()
56:	{
57:	$this->events = new TransportEvents();
58:	}
59:
60:	/**
61:	* @return array{self, self}
62:	*/
63:	public static function createPair(): array
64:	{
65:	$a = new self();
66:	$b = new self();
67:	$a->peer = $b;
68:	$b->peer = $a;
69:
70:	return [$a, $b];
71:	}
72:
73:	#[\Override]
74:	public function start(): void
75:	{
76:	match ($this->state) {
77:	TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
78:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
79:	TransportState::Idle => null,
80:	};
81:
82:	$this->state = TransportState::Running;
83:
84:	foreach ($this->pendingInbound as $envelope) {
85:	$this->deliver($envelope);
86:	}
87:	}
88:
89:	/**
90:	* `$context` is accepted for `TransportInterface` conformance and dropped, its `relatedRequestId` having
91:	* no in-process equivalent.
92:	*
93:	* @throws TransportAlreadyClosedException
94:	* @throws TransportNotStartedException
95:	*/
96:	#[\Override]
97:	public function send(JsonRpcMessage $message, ?SendContext $context = null): void
98:	{
99:	match ($this->state) {
100:	TransportState::Idle => throw new TransportNotStartedException(operation: 'send'),
101:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'send'),
102:	TransportState::Running => null,
103:	};
104:
105:	$this->peer?->receive($message->toArray());
106:	}
107:
108:	#[\Override]
109:	public function close(): void
110:	{
111:	if ($this->closing) {
112:	if (\Fiber::getCurrent() === $this->closingFiber) {
113:	return;
114:	}
115:
116:	$this->closeCompletion?->getFuture()->await();
117:
118:	return;
119:	}
120:
121:	$this->closing = true;
122:	$this->closingFiber = \Fiber::getCurrent();
123:
124:	/** @var DeferredFuture<null> $completion */
125:	$completion = new DeferredFuture();
126:	$this->closeCompletion = $completion;
127:
128:	try {
129:	$this->events->emitDrain();
130:	} finally {
131:	$this->state = TransportState::Closed;
132:
133:	try {
134:	$peer = $this->peer;
135:	$this->peer = null;
136:
137:	$peer?->close();
138:
139:	$this->events->emitClose();
140:	} finally {
141:	$completion->complete();
142:	}
143:	}
144:	}
145:
146:	#[\Override]
147:	public function onMessage(\Closure $listener): ListenerHandleInterface
148:	{
149:	return $this->events->onMessage($listener);
150:	}
151:
152:	/**
153:	* An in-memory transport has no I/O failure surface, so the error listeners see only faults thrown
154:	* by this side's own message listeners.
155:	*/
156:	#[\Override]
157:	public function onError(\Closure $listener): ListenerHandleInterface
158:	{
159:	return $this->events->onError($listener);
160:	}
161:
162:	#[\Override]
163:	public function onDrain(\Closure $listener): ListenerHandleInterface
164:	{
165:	return $this->events->onDrain($listener);
166:	}
167:
168:	#[\Override]
169:	public function onClose(\Closure $listener): ListenerHandleInterface
170:	{
171:	return $this->events->onClose($listener);
172:	}
173:
174:	/**
175:	* Cross-instance hand-off invoked by the peer's `send()`, queuing into `pendingInbound` while this side
176:	* is `Idle`.
177:	*
178:	* @param array<string, mixed> $envelope
179:	*/
180:	private function receive(array $envelope): void
181:	{
182:	if (TransportState::Idle === $this->state) {
183:	$this->pendingInbound[] = $envelope;
184:
185:	return;
186:	}
187:
188:	$this->deliver($envelope);
189:	}
190:
191:	/**
192:	* Emits an inbound envelope, keeping a listener fault on this side instead of surfacing it through
193:	* the peer's `send()`.
194:	*
195:	* @param array<string, mixed> $envelope
196:	*/
197:	private function deliver(array $envelope): void
198:	{
199:	try {
200:	$this->events->emitMessage($envelope, new ReceiveContext());
201:	} catch (\Throwable $e) {
202:	$this->events->emitError($e);
203:	}
204:	}
205:	}
206:

Namespaces

Classes

Interfaces

Enums