001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.juneau.rest.server.mcp;
018
019import java.util.Optional;
020
021/**
022 * Revision-neutral registry + filtered fan-out publisher for MCP {@code subscriptions/listen}.
023 *
024 * <p>
025 * Extends {@link McpSubscriptions} (the publish half) and adds registration/lifecycle: a v2 dispatch branch
026 * calls {@link #registerIfUnder(int, String, McpSubscriptionFilter)} once per accepted
027 * {@code subscriptions/listen} request. Close (graceful or abrupt) does <b>not</b> flow back through this
028 * interface: the returned {@link McpSubscription#close()} instead runs a close-callback wired at
029 * registration time, which removes the entry directly from the registry the returned handle's key was
030 * minted under. {@link #unregister(String)} is a registry-completeness API for that same by-id removal —
031 * it has no current production caller, since the close-callback path never has to look the id back up.
032 * Application handlers reach this same instance only through the narrower {@link McpSubscriptions} view.
033 *
034 * <p>
035 * Plain {@link #register(String, McpSubscriptionFilter)} is uncapped — it never rejects, so a caller wanting
036 * an enforced {@code maxConcurrentSubscriptions} ceiling must use {@link #registerIfUnder(int, String,
037 * McpSubscriptionFilter)} instead, which admission-checks and mutates the registry as one atomic step. A
038 * separate "check {@link #activeCount()}, then call {@code register}" pair is exactly the TOCTOU race this
039 * capped variant exists to close: two concurrent callers can both observe a count under the cap and both
040 * register before either mutation is visible to the other.
041 */
042public interface McpSubscriptionBroker extends McpSubscriptions {
043
044   /**
045    * Registers a new subscription and returns its queue handle. Uncapped: never rejects.
046    *
047    * @param subscriptionId The subscription identifier (the listen request's JSON-RPC id). Must not be
048    *    {@code null} or empty.
049    * @param honoredFilter The capability-gated filter this subscription honors. Must not be {@code null}.
050    * @return The new subscription handle. Never {@code null}.
051    */
052   McpSubscription register(String subscriptionId, McpSubscriptionFilter honoredFilter);
053
054   /**
055    * Atomically admission-checks against {@code max} and registers, closing the TOCTOU window a separate
056    * {@link #activeCount()} check plus {@link #register} call would leave open under concurrent callers
057    * (in particular against a JVM-wide shared broker instance, where unrelated callers race each other).
058    *
059    * <p>
060    * Re-registering an id that is already present is always allowed and never counted against {@code max},
061    * matching {@link #register}'s replace-on-same-id semantics (the previous subscription with that id is
062    * closed) — a replacement does not grow {@link #activeCount()}, so it cannot be what pushes the registry
063    * over the cap.
064    *
065    * @param max The maximum number of concurrently active subscriptions to admit under (exclusive: a
066    *    request is admitted only while {@link #activeCount()} is strictly less than this, except for the
067    *    same-id replacement case above).
068    * @param subscriptionId The subscription identifier (the listen request's JSON-RPC id). Must not be
069    *    {@code null} or empty.
070    * @param honoredFilter The capability-gated filter this subscription honors. Must not be {@code null}.
071    * @return The new subscription handle, or {@link Optional#empty()} if the cap was reached and this was
072    *    not a same-id replacement.
073    */
074   Optional<McpSubscription> registerIfUnder(int max, String subscriptionId, McpSubscriptionFilter honoredFilter);
075
076   /**
077    * Deregisters and closes the subscription with the given id, if present.
078    *
079    * @param subscriptionId The subscription identifier. A missing id is a no-op.
080    */
081   void unregister(String subscriptionId);
082
083   /**
084    * The number of currently registered (not yet closed) subscriptions.
085    *
086    * @return The active count.
087    */
088   int activeCount();
089
090   /**
091    * Closes every currently registered subscription.
092    *
093    * <p>
094    * Called when the server begins shutting down, so open {@code subscriptions/listen} streams complete
095    * promptly instead of holding the graceful drain open until its stop timeout.  The default implementation
096    * does nothing, so custom brokers that predate this method keep working (their streams then end on the
097    * normal heartbeat/idle-timeout paths).
098    */
099   default void closeAll() {
100      // No-op by default.
101   }
102}