Console Message Bus

The console message bus carries small JSON values between your server code and the pages of a console. The server publishes to a topic; every page that asked for the topic receives the value, and a page that connects later receives the last retained value straight away. Pages can also publish back on a few topics you allow. Use it to push live state (job counts, alerts, a terminal's new bytes) to a page without the page polling for it.

The bus is split over three opt-in modules. Add only what you need:

Module	Package	Adds
juneau-console-bus	org.apache.juneau.console.bus	ServerBus, BusPolicy, and the session POST and SSE stream (BusEventsMixin). Server to page only.
juneau-console-bus-websocket	org.apache.juneau.console.bus.websocket	The Jakarta WebSocket transport (BusWebSockets). The only transport that carries page to server frames.
juneau-terminal-bus	org.apache.juneau.terminal.bus	TerminalInput and TerminalNudges, which connect the terminal region to the bus.

juneau-console-bus depends on juneau-rest-server. The WebSocket module adds the bus module and the Jakarta WebSocket API; the servlet container supplies the implementation (for example Jetty's jetty-ee11-websocket-jakarta-server or Tomcat's tomcat-embed-websocket).

The pieces

Type	Role
ServerBus	Holds the retained topic values and the sessions that bridge them to pages. Create one per application and close() it when the application stops.
BusPolicy	Says what may flow, in which direction, to whom. Everything not allowed is denied.
BusEventsMixin	Interface you add to a REST resource. It exposes POST /juneau-bus/session and GET /juneau-bus/stream/{sessionId} relative to the resource.
UpstreamHandler	Handles one page-to-server topic.
BusSession	A connected page, as your code sees it: bridgeId(), transport(), principalName(), attribute(String), downstream(), upstream() and isConnected().
BusRefusal	The exception behind a refused session request; it carries the HTTP status and a bus:* code.

Publishing

A resource implements BusEventsMixin and returns the bus; the rest of the application publishes to the same bus.

@Rest(path="/ops/jobs")
public class JobsRest extends BasicRestServlet implements BusEventsMixin {

    static final ServerBus BUS = ServerBus.create(BusPolicy.create()
        .downstream("ops.jobs", true)       // retained state
        .downstream("ops.alert:*", false)   // events, any key
        .build());

    @Override
    public ServerBus serverBus() { return BUS; }
}

// Anywhere in the application:
JobsRest.BUS.publish("ops.jobs", JsonMap.of("schemaVersion", 1, "running", running));

Method	What it does
publish(topic, payload)	Sends to every connected session granted the topic. On a retained topic the value is also stored, and a value whose JSON equals the stored one is not sent again (returns false).
clear(topic)	Removes a retained value and tells the sessions granted the topic.
publishTo(audience, topic, payload)	Sends only to the sessions the Predicate<BusSession> accepts. Never retained. Returns how many sessions took the frame.
retainedJson(topic)	The stored JSON of a retained topic, as an Optional<String>.
sessions()	A snapshot of the live sessions.

Payloads are serialized as JSON. A publish to a topic the policy does not allow, or a frame larger than BusPolicy.maxFrameBytes(), throws IllegalArgumentException (E-56), so a typo fails loudly.

// Only to one user's pages; never retained.
BUS.publishTo(s -> "jb".equals(s.principalName()), "ops.jobs", mine);

Topics and the policy

Downstream topics are patterns the policy lists, each with a retain flag. A pattern is one of:

• a custom topic, ns.name (lower case letters, digits and -, a dot between the two parts);
• a keyed custom topic, ns.name:key, or every key of a family, ns.name:*;
• a job topic, job:<id> or job:*;
• a card command topic, cmd:<cardId>.

Upstream topics are exact custom topics, each with an UpstreamHandler.

BusPolicy policy = BusPolicy.create()
    .downstream("ops.jobs", true)
    .downstream("ops.alert:*", false)
    .upstream("ops.cancel-all", (session, topic, payload) -> jobs.cancelAll())
    .authorizer((req, topic) -> req.isUserInRole("ops"))
    .boundary(boundary)   // a LoopbackBoundary or CsrfBoundary
    .build();

The builder also sets the limits. The defaults are:

Setting	Default
sessionGrace	60 seconds after a session's last connection closes
maxSessionAge	12 hours
heartbeat	a ping frame after 15 seconds idle
maxSessions	256
maxQueuedFrames	1024 per connection
maxFrameBytes	65536, both directions
maxUpstreamPerSecond	20 per session

authorizer(BiPredicate<HttpServletRequest,String>) runs once per requested topic when the session opens, in both directions, and can only narrow a grant. sessionAttributes(Function<HttpServletRequest,Map<String,Object>>) captures values from the opening request that BusSession.attribute(String) returns later. boundary(RequestBoundary) puts the application's CSRF or loopback guard in front of the session POST.

Checking a page against the policy

A page asks for topics through the bridges in its page contract. In a page test, BusPolicy.check(contract) fails with an IllegalStateException that lists every requested topic the policy would deny (E-55), so a wiring mistake shows up in a test and not in the browser.

policy.check(PageCapture.of(html).contract());

How a page connects

1. The page POSTs to <resource>/juneau-bus/session with {"v":1, "bridge":"ops", "transport":"sse", "downstream":[...], "upstream":[...]}. bridge is required; transport is sse (the default) or websocket.
2. BusEventsMixin.openBusSession runs the policy boundary, then ServerBus opens a session and returns the grant: a 256-bit sessionId (the capability for the stream), the heartbeat and grace times, and the topics granted with their retain flags.
3. The page opens the stream. Over SSE that is GET <resource>/juneau-bus/stream/{sessionId}, a held-open text/event-stream with one event: bus per frame. Over WebSocket it is /juneau-bus/ws/{sessionId}.
4. The server sends the retained values as a resync between resync-begin and resync-end, then live frames.

A frame is one JSON object with v, type and, for the data types, topic, payload and a session seq. The types are pub, clear, resync-begin, resync-end, ping and error. BusFrames encodes and decodes them.

A refused session POST returns {"code":..., "message":..., "denied":[...]} with a status of 400 (bus:bad-request), 403 (bus:topic-denied), 429 (bus:too-many-sessions, with Retry-After), 501 (bus:transport-unavailable) or 503 (bus:closed). A stream request for an unknown or expired id, or from another principal, is a 404.

warning

The stream URL contains the session capability. Do not log request URIs for /juneau-bus/stream/*. Juneau's debug pipeline logs the request path at INFO or finer, and RestSession logs it at FINE on a client disconnect, so keep the loggers of resources that serve the bus coarser than INFO.

If the application has its own CSRF guard, override openBusSession in the resource, apply the guard, and call BusEventsMixin.super.openBusSession(req, res).

The WebSocket transport

SSE needs nothing beyond the bus module. Page-to-server frames need a WebSocket bridge. Register the endpoint once, when the web application starts:

public void contextInitialized(ServletContextEvent e) {
    BusWebSockets.register(e.getServletContext(), JobsRest.BUS);   // "/juneau-bus/ws/{sessionId}"
}

There is also a register(ServerContainer, ServerBus) form for code that holds the container, for example inside Jetty's JakartaWebSocketServletContainerInitializer.configure. Registration is strict and throws IllegalStateException rather than serving an unguarded endpoint:

• The container must expose a jakarta.websocket ServerContainer.
• The policy must have an Origin rule. That is LoopbackBoundary.origin() for a LoopbackBoundary, the handshake's own scheme and Host for a CsrfBoundary, or BusPolicy.allowedOrigins when the policy has no boundary.
• Any boundary type other than LoopbackBoundary and CsrfBoundary is refused, because a handshake cannot apply a boundary it does not know.

A handshake that fails the Host, Origin, Sec-Fetch-Site or capability checks is closed with 4403 (bus:refused) or 4401 (bus:unknown-session); a frame over maxFrameBytes is closed with 1009 (bus:frame-too-large). BusWebSocketConfigurator and BusWebSocketEndpoint are public only because the container instantiates them; do not construct them.

Upstream handlers

An UpstreamHandler receives a frame a page published on an upstream topic:

UpstreamHandler cancelAll = (session, topic, payload) -> jobs.running().forEach(AsyncJob::cancel);

The bus has already checked the session's upstream grant and that the payload is a JSON object. Handlers run on the container's WebSocket thread, so keep them short or hand the work off. An exception is logged and answered with a generic bus:upstream-failed frame; its text never crosses the wire. Frames above maxUpstreamPerSecond are refused.

Terminals on the bus

juneau-terminal-bus connects the terminal region to the bus. Both classes are built before the bus exists, then bound to it once.

TerminalInput is an UpstreamHandler for term.input. It passes a browser terminal's keystrokes to the terminal's input sink and answers each probe or refused frame on term.input-status, to the sending session only.

TerminalInput input = TerminalInput.create(access);
ServerBus bus = ServerBus.create(BusPolicy.create()
    .upstream(TerminalInput.TOPIC, input)
    .downstream(TerminalInput.STATUS_TOPIC, false)
    .build());
input.bind(bus);

TerminalNudges implements TerminalNudgePublisher. Hand it to TerminalProcess or MemoryTerminalSource and it publishes a small {"end": <offset>} hint on term.nudge:<id>, at most four per second per terminal, so a card fetches new bytes at once instead of waiting for its next poll. Its policy must allow term.nudge:* downstream without retention; bind throws otherwise.

TerminalNudges nudges = TerminalNudges.create(access);
ServerBus bus = ServerBus.create(BusPolicy.create()
    .downstream("term.nudge:*", false)
    .build());
nudges.bind(bus);

Both call TerminalAccess.canWatch for the caller, and TerminalInput also needs a present inputSink. The terminal region page covers the card side, the audit log, and the status states.