Namespaces

Classes

Interfaces

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\Subscription;
15:
16: use Amp\DeferredFuture;
17: use Nexus\Assert\Assert;
18: use Nexus\Mcp\Core\Handler\SenderInterface;
19: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcNotification;
20: use Nexus\Mcp\Core\Schema\MetaObject\NotificationMetaObject;
21: use Nexus\Mcp\Core\Schema\Notification\CancelledNotification;
22: use Nexus\Mcp\Core\Schema\Notification\PromptListChangedNotification;
23: use Nexus\Mcp\Core\Schema\Notification\ResourceListChangedNotification;
24: use Nexus\Mcp\Core\Schema\Notification\ResourceUpdatedNotification;
25: use Nexus\Mcp\Core\Schema\Notification\SubscriptionsAcknowledgedNotification;
26: use Nexus\Mcp\Core\Schema\Notification\ToolListChangedNotification;
27: use Nexus\Mcp\Core\Schema\NotificationParams\CancelledNotificationParams;
28: use Nexus\Mcp\Core\Schema\NotificationParams\EmptyNotificationParams;
29: use Nexus\Mcp\Core\Schema\NotificationParams\ResourceUpdatedNotificationParams;
30: use Nexus\Mcp\Core\Schema\NotificationParams\SubscriptionsAcknowledgedNotificationParams;
31: use Nexus\Mcp\Core\Schema\RequestId;
32: use Nexus\Mcp\Core\Schema\SubscriptionFilter;
33: use Nexus\Mcp\Server\Exception\SubscriptionLimitReachedException;
34: use Psr\Log\LoggerInterface;
35: use Psr\Log\NullLogger;
36: use Revolt\EventLoop;
37:
38: /**
39: * In-memory implementation of `SubscriptionStoreInterface`.
40: */
41: final class SubscriptionStore implements SubscriptionStoreInterface
42: {
43: public const int DEFAULT_MAX_SUBSCRIPTIONS = 1_024;
44: public const int DEFAULT_MAX_SUBSCRIPTIONS_PER_PEER = 256;
45: public const int DEFAULT_MAX_RESOURCE_SUBSCRIPTIONS_PER_STREAM = 256;
46:
47: /**
48: * @var array<int, SubscriptionEntry>
49: */
50: private array $entries = [];
51:
52: private bool $drained = false;
53: private int $pendingOpens = 0;
54:
55: /**
56: * Streams held per peer, pending acknowledgements included.
57: *
58: * @var array<non-empty-string, int>
59: */
60: private array $peerHeld = [];
61:
62: /**
63: * @var array<int, non-empty-string>
64: */
65: private array $entryPeers = [];
66:
67: /**
68: * @var array<string, array<int, SubscriptionEntry>>
69: */
70: private array $uriWatchers = [];
71:
72: /**
73: * @var array<non-empty-string, true>
74: */
75: private array $pendingListChanges = [];
76:
77: /**
78: * @var array<non-empty-string, true>
79: */
80: private array $pendingResourceUpdates = [];
81:
82: /**
83: * @param int<1, max> $maxSubscriptions
84: * @param int<1, max> $maxSubscriptionsPerPeer
85: * @param int<1, max> $maxResourceSubscriptionsPerStream
86: */
87: public function __construct(
88: private readonly bool $toolsListChanged = false,
89: private readonly bool $promptsListChanged = false,
90: private readonly bool $resourcesListChanged = false,
91: private readonly bool $resourceSubscriptions = false,
92: private readonly LoggerInterface $logger = new NullLogger(),
93: private readonly int $maxSubscriptions = self::DEFAULT_MAX_SUBSCRIPTIONS,
94: private readonly int $maxSubscriptionsPerPeer = self::DEFAULT_MAX_SUBSCRIPTIONS_PER_PEER,
95: private readonly int $maxResourceSubscriptionsPerStream = self::DEFAULT_MAX_RESOURCE_SUBSCRIPTIONS_PER_STREAM,
96: ) {
97: Assert::that($maxSubscriptions)->isPositiveInt(
98: 'The maximum open subscriptions must be a positive integer, {value} given.',
99: );
100: Assert::that($maxSubscriptionsPerPeer)->isPositiveInt(
101: 'The maximum open subscriptions per peer must be a positive integer, {value} given.',
102: );
103: Assert::that($maxResourceSubscriptionsPerStream)->isPositiveInt(
104: 'The maximum resource subscriptions per stream must be a positive integer, {value} given.',
105: );
106: }
107:
108: #[\Override]
109: public function open(RequestId $subscriptionId, SubscriptionFilter $requested, SenderInterface $sender, ?string $peer = null): SubscriptionEntry
110: {
111: $honoured = $this->honour($requested);
112:
113: if ($this->maxResourceSubscriptionsPerStream < \count($honoured->resourceSubscriptions ?? [])) {
114: throw new SubscriptionLimitReachedException($this->maxResourceSubscriptionsPerStream, $subscriptionId, perStream: true);
115: }
116:
117: if ($this->maxSubscriptions <= \count($this->entries) + $this->pendingOpens) {
118: throw new SubscriptionLimitReachedException($this->maxSubscriptions, $subscriptionId);
119: }
120:
121: if (null !== $peer) {
122: $held = $this->peerHeld[$peer] ?? 0;
123:
124: if ($this->maxSubscriptionsPerPeer <= $held) {
125: throw new SubscriptionLimitReachedException($this->maxSubscriptionsPerPeer, $subscriptionId, perPeer: true);
126: }
127:
128: $this->peerHeld[$peer] = $held + 1;
129: }
130:
131: /** @var DeferredFuture<null> $closed */
132: $closed = new DeferredFuture();
133: $entry = new SubscriptionEntry($subscriptionId, $honoured, $sender, $closed);
134:
135: // The acknowledgement can suspend, so the slot is held from the check until registration.
136: ++$this->pendingOpens;
137:
138: try {
139: $sender->sendNotification(new SubscriptionsAcknowledgedNotification(
140: params: new SubscriptionsAcknowledgedNotificationParams(
141: notifications: $honoured,
142: meta: new NotificationMetaObject(subscriptionId: $subscriptionId),
143: ),
144: ));
145: } catch (\Throwable $e) {
146: if (null !== $peer) {
147: $this->releasePeer($peer);
148: }
149:
150: throw $e;
151: } finally {
152: --$this->pendingOpens;
153: }
154:
155: if ($this->drained) {
156: if (null !== $peer) {
157: $this->releasePeer($peer);
158: }
159:
160: $closed->complete();
161:
162: return $entry;
163: }
164:
165: $id = spl_object_id($entry);
166: $this->entries[$id] = $entry;
167:
168: if (null !== $peer) {
169: $this->entryPeers[$id] = $peer;
170: }
171:
172: foreach ($honoured->resourceSubscriptions ?? [] as $uri) {
173: $this->uriWatchers[$uri][$id] = $entry;
174: }
175:
176: return $entry;
177: }
178:
179: #[\Override]
180: public function close(SubscriptionEntry $entry): void
181: {
182: if (! \array_key_exists(spl_object_id($entry), $this->entries)) {
183: return;
184: }
185:
186: $this->discard($entry);
187:
188: $this->pushTo($entry, new CancelledNotification(
189: params: new CancelledNotificationParams(
190: requestId: $entry->subscriptionId,
191: meta: new NotificationMetaObject(subscriptionId: $entry->subscriptionId),
192: ),
193: ));
194:
195: $entry->closed->complete();
196: }
197:
198: #[\Override]
199: public function discard(SubscriptionEntry $entry): void
200: {
201: $id = spl_object_id($entry);
202:
203: if (isset($this->entryPeers[$id])) {
204: $this->releasePeer($this->entryPeers[$id]);
205: unset($this->entryPeers[$id]);
206: }
207:
208: foreach ($entry->honoured->resourceSubscriptions ?? [] as $uri) {
209: unset($this->uriWatchers[$uri][$id]);
210:
211: if ([] === ($this->uriWatchers[$uri] ?? [])) {
212: unset($this->uriWatchers[$uri]);
213: }
214: }
215:
216: unset($this->entries[$id]);
217: }
218:
219: #[\Override]
220: public function closeAll(): void
221: {
222: $this->drained = true;
223:
224: foreach ($this->entries as $entry) {
225: $this->close($entry);
226: }
227: }
228:
229: #[\Override]
230: public function reopen(): void
231: {
232: $this->drained = false;
233: }
234:
235: #[\Override]
236: public function emitToolListChanged(): void
237: {
238: $this->pendingListChanges['tools'] = true;
239: $this->scheduleEndOfTickFlush();
240: }
241:
242: #[\Override]
243: public function emitPromptListChanged(): void
244: {
245: $this->pendingListChanges['prompts'] = true;
246: $this->scheduleEndOfTickFlush();
247: }
248:
249: #[\Override]
250: public function emitResourceListChanged(): void
251: {
252: $this->pendingListChanges['resources'] = true;
253: $this->scheduleEndOfTickFlush();
254: }
255:
256: #[\Override]
257: public function emitResourceUpdated(string $uri): void
258: {
259: Assert::that($uri)->isNonEmptyString('An updated resource URI must be a non-empty string.');
260:
261: $this->pendingResourceUpdates[$uri] = true;
262: $this->scheduleEndOfTickFlush();
263: }
264:
265: #[\Override]
266: public function honour(SubscriptionFilter $requested): SubscriptionFilter
267: {
268: return new SubscriptionFilter(
269: toolsListChanged: $this->toolsListChanged && true === $requested->toolsListChanged ? true : null,
270: promptsListChanged: $this->promptsListChanged && true === $requested->promptsListChanged ? true : null,
271: resourcesListChanged: $this->resourcesListChanged && true === $requested->resourcesListChanged ? true : null,
272: resourceSubscriptions: $this->resourceSubscriptions ? $requested->resourceSubscriptions : null,
273: );
274: }
275:
276: /**
277: * @param non-empty-string $peer
278: */
279: private function releasePeer(string $peer): void
280: {
281: $held = $this->peerHeld[$peer] - 1;
282:
283: if ($held < 1) {
284: unset($this->peerHeld[$peer]);
285: } else {
286: $this->peerHeld[$peer] = $held;
287: }
288: }
289:
290: private function scheduleEndOfTickFlush(): void
291: {
292: EventLoop::defer(function (): void {
293: $kinds = $this->pendingListChanges;
294: $uris = $this->pendingResourceUpdates;
295: $this->pendingListChanges = [];
296: $this->pendingResourceUpdates = [];
297:
298: foreach (['tools', 'prompts', 'resources'] as $kind) {
299: if (\array_key_exists($kind, $kinds)) {
300: $this->broadcastListChange($kind);
301: }
302: }
303:
304: foreach (array_keys($uris) as $uri) {
305: $this->broadcastResourceUpdate((string) $uri);
306: }
307: });
308: }
309:
310: /**
311: * @param non-empty-string $kind
312: */
313: private function broadcastListChange(string $kind): void
314: {
315: foreach ($this->entries as $entry) {
316: $honoured = $entry->honoured;
317: $wanted = match ($kind) {
318: 'tools' => true === $honoured->toolsListChanged,
319: 'prompts' => true === $honoured->promptsListChanged,
320: default => true === $honoured->resourcesListChanged,
321: };
322:
323: if (! $wanted) {
324: continue;
325: }
326:
327: $params = new EmptyNotificationParams(new NotificationMetaObject(subscriptionId: $entry->subscriptionId));
328: $this->pushTo($entry, match ($kind) {
329: 'tools' => new ToolListChangedNotification(params: $params),
330: 'prompts' => new PromptListChangedNotification(params: $params),
331: default => new ResourceListChangedNotification(params: $params),
332: });
333: }
334: }
335:
336: /**
337: * @param non-empty-string $uri
338: */
339: private function broadcastResourceUpdate(string $uri): void
340: {
341: foreach ($this->uriWatchers[$uri] ?? [] as $entry) {
342: $this->pushTo($entry, new ResourceUpdatedNotification(
343: params: new ResourceUpdatedNotificationParams(
344: uri: $uri,
345: meta: new NotificationMetaObject(subscriptionId: $entry->subscriptionId),
346: ),
347: ));
348: }
349: }
350:
351: /**
352: * Sends one notification to one stream, so a failure or a mid-broadcast teardown cannot cost the
353: * streams behind it.
354: *
355: * @param JsonRpcNotification<non-empty-string> $notification
356: */
357: private function pushTo(SubscriptionEntry $entry, JsonRpcNotification $notification): void
358: {
359: if ($entry->closed->isComplete()) {
360: return;
361: }
362:
363: try {
364: $entry->sender->sendNotification($notification);
365: } catch (\Throwable $e) {
366: $this->logger->debug('Dropping a subscription notification its stream could not take.', [
367: 'method' => $notification::getMethod(),
368: 'exception' => $e,
369: ]);
370: }
371: }
372: }
373:
Nexus MCP SDK (1.x-dev) API documentation generated by ApiGen dev-master