File Server/Subscription/SubscriptionStore.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\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:	/**
44:	* How many streams one store holds open before it starts refusing new ones.
45:	*/
46:	public const int DEFAULT_MAX_SUBSCRIPTIONS = 1024;
47:
48:	private const string TOOLS = 'tools';
49:	private const string PROMPTS = 'prompts';
50:	private const string RESOURCES = 'resources';
51:
52:	/**
53:	* Keyed by object identity, not by any request id. Request ids are unique per connection, and a
54:	* sessionless endpoint serves many connections through one store.
55:	*
56:	* @var array<int, SubscriptionEntry>
57:	*/
58:	private array $entries = [];
59:
60:	private bool $drained = false;
61:
62:	/**
63:	* List-change kinds, and resource URIs, awaiting the end of the current event-loop tick.
64:	*
65:	* @var array<non-empty-string, true>
66:	*/
67:	private array $pendingListChanges = [];
68:
69:	/**
70:	* @var array<non-empty-string, true>
71:	*/
72:	private array $pendingResourceUpdates = [];
73:
74:	public function __construct(
75:	private readonly bool $toolsListChanged = false,
76:	private readonly bool $promptsListChanged = false,
77:	private readonly bool $resourcesListChanged = false,
78:	private readonly bool $resourceSubscriptions = false,
79:	private readonly LoggerInterface $logger = new NullLogger(),
80:	private readonly int $maxSubscriptions = self::DEFAULT_MAX_SUBSCRIPTIONS,
81:	) {
82:	Assert::that($maxSubscriptions)
83:	->isPositiveInt('The maximum open subscriptions must be a positive integer, {value} given.')
84:	;
85:	}
86:
87:	#[\Override]
88:	public function open(RequestId $subscriptionId, SubscriptionFilter $requested, SenderInterface $sender): SubscriptionEntry
89:	{
90:	if ($this->maxSubscriptions <= \count($this->entries)) {
91:	// Refused before the acknowledgement, so the client never sees a stream it does not have.
92:	throw new SubscriptionLimitReachedException($this->maxSubscriptions, $subscriptionId);
93:	}
94:
95:	$honoured = $this->honour($requested);
96:
97:	/** @var DeferredFuture<null> $closed */
98:	$closed = new DeferredFuture();
99:	$entry = new SubscriptionEntry($subscriptionId, $honoured, $sender, $closed);
100:
101:	// The spec makes the acknowledgement the first message on the stream, so it goes out before the
102:	// entry is visible to any emit.
103:	$sender->sendNotification(new SubscriptionsAcknowledgedNotification(
104:	params: new SubscriptionsAcknowledgedNotificationParams(
105:	notifications: $honoured,
106:	meta: new NotificationMetaObject(subscriptionId: $subscriptionId),
107:	),
108:	));
109:
110:	if ($this->drained) {
111:	// The server is shutting down. Settle the stream at once so its handler cannot outlive the drain.
112:	$closed->complete();
113:
114:	return $entry;
115:	}
116:
117:	$this->entries[spl_object_id($entry)] = $entry;
118:
119:	return $entry;
120:	}
121:
122:	#[\Override]
123:	public function close(SubscriptionEntry $entry): void
124:	{
125:	if (! \array_key_exists(spl_object_id($entry), $this->entries)) {
126:	return;
127:	}
128:
129:	$this->discard($entry);
130:
131:	// The spec has the server name the `subscriptions/listen` request it is tearing down, and tags
132:	// every notification delivered on a stream with that stream's id.
133:	$this->pushTo($entry, new CancelledNotification(
134:	params: new CancelledNotificationParams(
135:	requestId: $entry->subscriptionId,
136:	meta: new NotificationMetaObject(subscriptionId: $entry->subscriptionId),
137:	),
138:	));
139:
140:	$entry->closed->complete();
141:	}
142:
143:	#[\Override]
144:	public function discard(SubscriptionEntry $entry): void
145:	{
146:	unset($this->entries[spl_object_id($entry)]);
147:	}
148:
149:	#[\Override]
150:	public function closeAll(): void
151:	{
152:	$this->drained = true;
153:
154:	foreach ($this->entries as $entry) {
155:	$this->close($entry);
156:	}
157:	}
158:
159:	#[\Override]
160:	public function emitToolListChanged(): void
161:	{
162:	$this->coalesceListChange(self::TOOLS);
163:	}
164:
165:	#[\Override]
166:	public function emitPromptListChanged(): void
167:	{
168:	$this->coalesceListChange(self::PROMPTS);
169:	}
170:
171:	#[\Override]
172:	public function emitResourceListChanged(): void
173:	{
174:	$this->coalesceListChange(self::RESOURCES);
175:	}
176:
177:	#[\Override]
178:	public function emitResourceUpdated(string $uri): void
179:	{
180:	Assert::that($uri)->isNonEmptyString('An updated resource URI must be a non-empty string.');
181:
182:	$this->pendingResourceUpdates[$uri] = true;
183:	$this->scheduleFlush();
184:	}
185:
186:	#[\Override]
187:	public function honour(SubscriptionFilter $requested): SubscriptionFilter
188:	{
189:	return new SubscriptionFilter(
190:	toolsListChanged: $this->toolsListChanged && true === $requested->toolsListChanged ? true : null,
191:	promptsListChanged: $this->promptsListChanged && true === $requested->promptsListChanged ? true : null,
192:	resourcesListChanged: $this->resourcesListChanged && true === $requested->resourcesListChanged ? true : null,
193:	resourceSubscriptions: $this->resourceSubscriptions ? $requested->resourceSubscriptions : null,
194:	);
195:	}
196:
197:	/**
198:	* @param self::PROMPTS\|self::RESOURCES\|self::TOOLS $kind
199:	*/
200:	private function coalesceListChange(string $kind): void
201:	{
202:	$this->pendingListChanges[$kind] = true;
203:	$this->scheduleFlush();
204:	}
205:
206:	/**
207:	* Holds announcements until the end of the tick, so a burst of mutations reaches each stream once.
208:	*/
209:	private function scheduleFlush(): void
210:	{
211:	EventLoop::defer(function (): void {
212:	$kinds = $this->pendingListChanges;
213:	$uris = $this->pendingResourceUpdates;
214:	$this->pendingListChanges = [];
215:	$this->pendingResourceUpdates = [];
216:
217:	foreach ([self::TOOLS, self::PROMPTS, self::RESOURCES] as $kind) {
218:	if (\array_key_exists($kind, $kinds)) {
219:	$this->broadcastListChange($kind);
220:	}
221:	}
222:
223:	foreach (array_keys($uris) as $uri) {
224:	// An all-digit URI is a legal string but an int array key, so it comes back coerced.
225:	$this->broadcastResourceUpdate((string) $uri);
226:	}
227:	});
228:	}
229:
230:	/**
231:	* @param non-empty-string $kind
232:	*/
233:	private function broadcastListChange(string $kind): void
234:	{
235:	foreach ($this->entries as $entry) {
236:	$honoured = $entry->honoured;
237:	$wanted = match ($kind) {
238:	self::TOOLS => true === $honoured->toolsListChanged,
239:	self::PROMPTS => true === $honoured->promptsListChanged,
240:	default => true === $honoured->resourcesListChanged,
241:	};
242:
243:	if (! $wanted) {
244:	continue;
245:	}
246:
247:	$params = new EmptyNotificationParams(new NotificationMetaObject(subscriptionId: $entry->subscriptionId));
248:	$this->pushTo($entry, match ($kind) {
249:	self::TOOLS => new ToolListChangedNotification(params: $params),
250:	self::PROMPTS => new PromptListChangedNotification(params: $params),
251:	default => new ResourceListChangedNotification(params: $params),
252:	});
253:	}
254:	}
255:
256:	/**
257:	* @param non-empty-string $uri
258:	*/
259:	private function broadcastResourceUpdate(string $uri): void
260:	{
261:	foreach ($this->entries as $entry) {
262:	if (! \in_array($uri, $entry->honoured->resourceSubscriptions ?? [], true)) {
263:	continue;
264:	}
265:
266:	$this->pushTo($entry, new ResourceUpdatedNotification(
267:	params: new ResourceUpdatedNotificationParams(
268:	uri: $uri,
269:	meta: new NotificationMetaObject(subscriptionId: $entry->subscriptionId),
270:	),
271:	));
272:	}
273:	}
274:
275:	/**
276:	* Sends one notification to one stream. A send that fails must not cost the streams behind it, and a
277:	* stream torn down mid-broadcast must hear nothing further.
278:	*
279:	* @param JsonRpcNotification<non-empty-string> $notification
280:	*/
281:	private function pushTo(SubscriptionEntry $entry, JsonRpcNotification $notification): void
282:	{
283:	if ($entry->closed->isComplete()) {
284:	return;
285:	}
286:
287:	try {
288:	$entry->sender->sendNotification($notification);
289:	} catch (\Throwable $e) {
290:	$this->logger->debug('Dropping a subscription notification its stream could not take.', [
291:	'method' => $notification::getMethod(),
292:	'exception' => $e,
293:	]);
294:	}
295:	}
296:	}
297:

Namespaces

Classes

Interfaces