Package org.apache.juneau.rest.server.mcp

Interface McpSubscriptionBroker

All Superinterfaces:
McpSubscriptions
All Known Implementing Classes:
BasicMcpSubscriptionBroker
public interface McpSubscriptionBroker extends McpSubscriptions
Revision-neutral registry + filtered fan-out publisher for MCP subscriptions/listen.

Extends McpSubscriptions (the publish half) and adds registration/lifecycle: a v2 dispatch branch calls registerIfUnder(int, String, McpSubscriptionFilter) once per accepted subscriptions/listen request. Close (graceful or abrupt) does not flow back through this interface: the returned McpSubscription.close() instead runs a close-callback wired at registration time, which removes the entry directly from the registry the returned handle's key was minted under. unregister(String) is a registry-completeness API for that same by-id removal — it has no current production caller, since the close-callback path never has to look the id back up. Application handlers reach this same instance only through the narrower McpSubscriptions view.

Plain register(String, McpSubscriptionFilter) is uncapped — it never rejects, so a caller wanting an enforced maxConcurrentSubscriptions ceiling must use registerIfUnder(int, String, McpSubscriptionFilter) instead, which admission-checks and mutates the registry as one atomic step. A separate "check activeCount(), then call register" pair is exactly the TOCTOU race this capped variant exists to close: two concurrent callers can both observe a count under the cap and both register before either mutation is visible to the other.

  • Method Details

    • register

      McpSubscription register(String subscriptionId, McpSubscriptionFilter honoredFilter)
      Registers a new subscription and returns its queue handle. Uncapped: never rejects.
      Parameters:
      subscriptionId - The subscription identifier (the listen request's JSON-RPC id). Must not be null or empty.
      honoredFilter - The capability-gated filter this subscription honors. Must not be null.
      Returns:
      The new subscription handle. Never null.

    • registerIfUnder

      Optional<McpSubscription> registerIfUnder(int max, String subscriptionId, McpSubscriptionFilter honoredFilter)
      Atomically admission-checks against max and registers, closing the TOCTOU window a separate activeCount() check plus register(java.lang.String, org.apache.juneau.rest.server.mcp.McpSubscriptionFilter) call would leave open under concurrent callers (in particular against a JVM-wide shared broker instance, where unrelated callers race each other).

      Re-registering an id that is already present is always allowed and never counted against max, matching register(java.lang.String, org.apache.juneau.rest.server.mcp.McpSubscriptionFilter)'s replace-on-same-id semantics (the previous subscription with that id is closed) — a replacement does not grow activeCount(), so it cannot be what pushes the registry over the cap.

      Parameters:
      max - The maximum number of concurrently active subscriptions to admit under (exclusive: a request is admitted only while activeCount() is strictly less than this, except for the same-id replacement case above).
      subscriptionId - The subscription identifier (the listen request's JSON-RPC id). Must not be null or empty.
      honoredFilter - The capability-gated filter this subscription honors. Must not be null.
      Returns:
      The new subscription handle, or Optional.empty() if the cap was reached and this was not a same-id replacement.

    • unregister

      void unregister(String subscriptionId)
      Deregisters and closes the subscription with the given id, if present.
      Parameters:
      subscriptionId - The subscription identifier. A missing id is a no-op.

    • activeCount

      int activeCount()
      The number of currently registered (not yet closed) subscriptions.
      Returns:
      The active count.

    • closeAll

      default void closeAll()
      Closes every currently registered subscription.

      Called when the server begins shutting down, so open subscriptions/listen streams complete promptly instead of holding the graceful drain open until its stop timeout. The default implementation does nothing, so custom brokers that predate this method keep working (their streams then end on the normal heartbeat/idle-timeout paths).