Namespaces

Classes

Interfaces

Enums

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 Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
17: use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
18: use Nexus\Mcp\Core\Exception\TransportNotStartedException;
19: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
20:
21: /**
22: * In-process JSON-RPC duplex between two `TransportInterface` instances. Each
23: * side's `send()` becomes the other side's inbound message. Pre-start inbound
24: * envelopes queue and drain on `start()`. `close()` cascades to the peer.
25: */
26: final class InMemoryTransport implements TransportInterface
27: {
28: /**
29: * Envelopes the peer's `send()` delivered before this side called `start()`. Drained in arrival order on `start()`.
30: *
31: * @var list<array<string, mixed>>
32: */
33: private array $pendingInbound = [];
34:
35: private TransportState $state = TransportState::Idle;
36: private ?self $peer = null;
37: private readonly TransportEvents $events;
38:
39: private function __construct()
40: {
41: $this->events = new TransportEvents();
42: }
43:
44: /**
45: * Returns two linked transports. Each side's `send()` delivers to the
46: * other side's `onMessage` listeners. Use one for the server, the other
47: * for the client.
48: *
49: * @return array{self, self}
50: */
51: public static function createPair(): array
52: {
53: $a = new self();
54: $b = new self();
55: $a->peer = $b;
56: $b->peer = $a;
57:
58: return [$a, $b];
59: }
60:
61: #[\Override]
62: public function start(): void
63: {
64: match ($this->state) {
65: TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
66: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
67: TransportState::Idle => null,
68: };
69:
70: $this->state = TransportState::Running;
71:
72: foreach ($this->pendingInbound as $envelope) {
73: $this->events->emitMessage($envelope, new ReceiveContext());
74: }
75: }
76:
77: /**
78: * `$context`'s `relatedRequestId` is a streamable-HTTP concern with no in-process
79: * equivalent. The parameter is accepted for `TransportInterface` conformance and
80: * intentionally dropped.
81: *
82: * @throws TransportAlreadyClosedException
83: * @throws TransportNotStartedException
84: */
85: #[\Override]
86: public function send(JsonRpcMessage $message, ?SendContext $context = null): void
87: {
88: match ($this->state) {
89: TransportState::Idle => throw new TransportNotStartedException(operation: 'send'),
90: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'send'),
91: TransportState::Running => null,
92: };
93:
94: $this->peer?->receive($message->toArray());
95: }
96:
97: #[\Override]
98: public function close(): void
99: {
100: if (TransportState::Closed === $this->state) {
101: return;
102: }
103:
104: try {
105: $this->events->emitDrain();
106: } finally {
107: $this->state = TransportState::Closed;
108:
109: $peer = $this->peer;
110: $this->peer = null;
111:
112: $peer?->close();
113:
114: $this->events->emitClose();
115: }
116: }
117:
118: #[\Override]
119: public function onMessage(\Closure $listener): SubscriptionInterface
120: {
121: return $this->events->onMessage($listener);
122: }
123:
124: /**
125: * In-memory transport has no I/O failure surface, so registered error
126: * listeners are accepted for contract conformance but never invoked.
127: */
128: #[\Override]
129: public function onError(\Closure $listener): SubscriptionInterface
130: {
131: return $this->events->onError($listener);
132: }
133:
134: #[\Override]
135: public function onDrain(\Closure $listener): SubscriptionInterface
136: {
137: return $this->events->onDrain($listener);
138: }
139:
140: #[\Override]
141: public function onClose(\Closure $listener): SubscriptionInterface
142: {
143: return $this->events->onClose($listener);
144: }
145:
146: /**
147: * Cross-instance hand-off invoked by the peer's `send()`. Queues into
148: * `pendingInbound` while this side is `Idle`, otherwise emits to listeners.
149: *
150: * @param array<string, mixed> $envelope
151: */
152: private function receive(array $envelope): void
153: {
154: if (TransportState::Idle === $this->state) {
155: $this->pendingInbound[] = $envelope;
156:
157: return;
158: }
159:
160: $this->events->emitMessage($envelope, new ReceiveContext());
161: }
162: }
163:
Nexus MCP SDK API documentation generated by ApiGen dev-master