Package org.apache.juneau.console.bus

Class ServerBus

java.lang.Object
org.apache.juneau.console.bus.ServerBus
All Implemented Interfaces:
AutoCloseable
public final class ServerBus extends Object implements AutoCloseable
The server half of the console message bus: a retained-value store plus the sessions that bridge it to pages.

Publish from anywhere in the application; every connected page whose bridge was granted the topic receives it (and every page that connects later receives the retained value). Topics not allowed by the BusPolicy are refused loudly (E-56).

The methods under "transport SPI" are public for BusEventsMixin and BusWebSockets; application code does not call them.

Example:

// One per application (or per resource), shared by the REST resource and any background work. ServerBus bus = ServerBus.create(BusPolicy.create() .downstream("ops.jobs", true) .downstream("cmd:jobs", false) .build()); bus.publish("ops.jobs", JsonMap.of("schemaVersion", 1, "running", running)); bus.publish("cmd:jobs", JsonMap.of("schemaVersion", 1, "op", "reload")); // Only to one user's pages (never retained): bus.publishTo(s -> "jb".equals(s.principalName()), "ops.jobs", mine);

Since:
10.0.0

  • Method Details

    • create

      public static ServerBus create(BusPolicy policy)
      Creates a bus and starts its timer (one daemon thread: heartbeat pings and session expiry).
      Parameters:
      policy - The policy. Must not be null.
      Returns:
      A new bus. close() it when the application stops.

    • publish

      public boolean publish(String topic, Object payload)
      Publishes a value to every connected session granted the topic, and stores it if the topic is retained.

      For a retained topic, a value whose JSON equals the stored JSON is not republished (distinct-until-changed).

      Parameters:
      topic - A concrete topic the policy allows downstream (not a :* pattern).
      payload - The payload, serialized with Json.of(Object). Must not be null; use clear(String) to remove a retained value.
      Returns:
      false when the topic is retained and the value is unchanged; otherwise true.
      Throws:
      IllegalArgumentException - E-56 when the policy denies the topic or the frame is larger than BusPolicy.maxFrameBytes().

    • clear

      public void clear(String topic)
      Removes a retained value and sends clear to every connected session granted the topic.

      Clearing a topic that holds no value does nothing.

      Parameters:
      topic - A retained topic.
      Throws:
      IllegalArgumentException - E-56 when the policy denies the topic or does not retain it.

    • publishTo

      public int publishTo(Predicate<BusSession> audience, String topic, Object payload)
      Publishes a value to the connected sessions that match an audience. Never retained, even on a retained topic.
      Parameters:
      audience - Selects sessions. Runs under the bus lock: keep it cheap and side-effect free.
      topic - A concrete topic the policy allows downstream.
      payload - The payload. Must not be null.
      Returns:
      The number of sessions that accepted the frame.
      Throws:
      IllegalArgumentException - E-56, as for publish(String, Object).

    • retainedJson

      public Optional<String> retainedJson(String topic)
      The stored JSON of a retained topic.
      Parameters:
      topic - The topic.
      Returns:
      The JSON, or empty when nothing is stored.

    • sessions

      public List<BusSession> sessions()
      The live sessions (connected, or disconnected within their grace period).
      Returns:
      An immutable snapshot.

    • policy

      public BusPolicy policy()
      Returns:
      The policy this bus enforces.

    • close

      public void close()
      Closes every connection with 1001 (bus:closed), forgets every session and stops the timer. Retained values are kept. Idempotent.
      Specified by:
      close in interface AutoCloseable

    • openSession

      public BusSession openSession(jakarta.servlet.http.HttpServletRequest req, JsonMap body) throws BusRefusal
      Transport SPI: opens a session from a session POST body.

      Body: {"v":1, "bridge":"ops", "transport":"sse"|"websocket", "downstream":[...], "upstream":[...]}. v defaults to 1 and transport to "sse"; bridge is required.

      Parameters:
      req - The POST request: the authorizer's argument and the source of the principal and session attributes.
      body - The parsed body. null is a 400.
      Returns:
      The new session.
      Throws:
      BusRefusal - 400 bus:bad-request, 403 bus:topic-denied, 429 bus:too-many-sessions, 501 bus:transport-unavailable, or 503 bus:closed.

    • session

      public Optional<BusSession> session(String id, Principal principal)
      Transport SPI: finds a live session by its capability id, for the principal that opened it.
      Parameters:
      id - The session id from the request path. Can be null.
      principal - The request principal. Can be null (anonymous).
      Returns:
      The session, or empty when the id is unknown or expired, or the principal differs. The three cases are deliberately indistinguishable.

    • connect

      public void connect(BusSession session, FrameSink sink)
      Transport SPI: attaches a connection to a session, then sends the resync.

      Any live sink is replaced and closed with 4409 (bus:replaced). The resync is resync-begin, one retained pub for every stored value the session's grant covers, in topic order, then resync-end. When the bus is closed or the session has expired, the sink is closed at once (1001 / 4401).

      Parameters:
      session - A session of this bus.
      sink - The connection's outbound side.

    • disconnect

      public void disconnect(BusSession session, FrameSink sink)
      Transport SPI: the connection behind a sink has closed. Starts the session's grace period.

      A sink that is no longer the session's live sink (it was replaced, or detached as a slow consumer) is ignored.

      Parameters:
      session - A session of this bus.
      sink - The sink passed to connect(BusSession, FrameSink).

    • receive

      public void receive(BusSession session, String frameText)
      Transport SPI: one upstream frame from a WebSocket connection.

      Refusals are answered with an error frame and the frame is dropped; the connection stays open.

      Parameters:
      session - A session of this bus.
      frameText - The frame text.

    • enableWebSocket

      public void enableWebSocket(String contextRelativePathPrefix)
      Transport SPI: records that the WebSocket transport is registered, so transport:"websocket" sessions may open and the session POST response carries paths.websocket.
      Parameters:
      contextRelativePathPrefix - The endpoint path without its /{sessionId} segment, for example /juneau-bus/ws. Must start with / and not end with /.