Psr16NotificationBus implements NotificationBusInterface
A notification bus over any PSR-16 cache, so publishers and listen streams in different processes can see each other.
The sequence counter and the entries are separate keys: a publisher bumps the
counter and writes its entry under the number it got. That is not atomic
across processes — two publishers can take the same number and one entry is
lost — which is the trade every PSR-16-only design makes, and acceptable
here: a lost tools/list_changed costs a client one stale list until the TTL
lapses, not correctness. A backend with atomic increments (Redis INCR)
should implement NotificationBusInterface directly.
Tags
Table of Contents
Interfaces
- NotificationBusInterface
- Carries server-initiated notifications to whichever `subscriptions/listen` streams are open.
Constants
- CURSOR_KEY : mixed = 'cursor'
Properties
- $backlog : int
- $cache : CacheInterface
- $logger : LoggerInterface
- $messageFactory : MessageFactory
- $prefix : string
- $ttl : int
Methods
- __construct() : mixed
- cursor() : int
- The cursor a stream opening now should start from.
- publish() : void
- Publishes a notification to every stream currently reading forward.
- since() : Notification>, int}
- Notifications published after $cursor, and the cursor to read from next.
- key() : string
Constants
CURSOR_KEY
private
mixed
CURSOR_KEY
= 'cursor'
Properties
$backlog read-only
private
int
$backlog
= 256
$cache read-only
private
CacheInterface
$cache
$logger read-only
private
LoggerInterface
$logger
= new NullLogger()
$messageFactory read-only
private
MessageFactory
$messageFactory
$prefix read-only
private
string
$prefix
= 'mcp.notifications.'
$ttl read-only
private
int
$ttl
= 120
Methods
__construct()
public
__construct(CacheInterface $cache[, string $prefix = 'mcp.notifications.' ][, int $ttl = 120 ][, int $backlog = 256 ][, LoggerInterface $logger = new NullLogger() ][, MessageFactory|null $messageFactory = null ]) : mixed
Parameters
- $cache : CacheInterface
- $prefix : string = 'mcp.notifications.'
-
namespace for this bus's keys, so one cache can carry several
- $ttl : int = 120
-
how long an entry stays readable, in seconds
- $backlog : int = 256
-
how many entries a reader will look back over
- $logger : LoggerInterface = new NullLogger()
- $messageFactory : MessageFactory|null = null
cursor()
The cursor a stream opening now should start from.
public
cursor() : int
Deliberately "now" and not "the beginning": a client that subscribes wants what happens next, not a replay of everything the server has done.
Return values
intpublish()
Publishes a notification to every stream currently reading forward.
public
publish(Notification $notification) : void
Parameters
- $notification : Notification
since()
Notifications published after $cursor, and the cursor to read from next.
public
since(int $cursor) : Notification>, int}
Parameters
- $cursor : int
Return values
Notification>, int}key()
private
key(string $suffix) : string
Parameters
- $suffix : string