Package org.apache.juneau.console.bus

Interface BusEventsMixin

public interface BusEventsMixin
Exposes the bus's session POST and SSE stream on a REST resource.
  • POST <servlet>/juneau-bus/session — opens a session and returns its grant. This is the protocol's only state-changing request, so every guard the application has applies to it.
  • GET <servlet>/juneau-bus/stream/{sessionId} — a held-open text/event-stream of bus frames, each as event: bus. Gated by the capability in the path; an unknown or expired id, or another principal, is a 404.

When BusPolicy.boundary() is set, both endpoints run its check: the full write check on the POST, and on the GET whatever the boundary applies to a read (a LoopbackBoundary's Host check, a CsrfBoundary's allowed hosts). The method judged is the one the request claims, so a GET carrying a method parameter or an override header that claims a write gets the write check.

Example:

@Rest(path="/ops/jobs") public class JobsRest extends BasicRestServlet implements AsyncJobsMixin, BusEventsMixin { static final ServerBus BUS = ServerBus.create(BusPolicy.create() .downstream("ops.jobs", true).build()); @Override public ServerBus serverBus() { return BUS; } ... }

Example:

// An application with its own CSRF guard puts it in front of the session POST. // Re-declaring @RestPost here is optional and registers the route once. @Override @RestPost(path=SESSION_PATH) public JsonMap openBusSession(RestRequest req, RestResponse res) { requireCsrfToken(req); return BusEventsMixin.super.openBusSession(req, res); }

Since:
10.0.0

  • Field Details

  • Method Details

    • serverBus

      ServerBus serverBus()
      The bus this resource serves.
      Returns:
      The bus. Typically one per application, shared with the code that publishes.

    • openBusSession

      default JsonMap openBusSession(RestRequest req, RestResponse res)
      [POST /juneau-bus/session] — opens a bus session.

      Body (at most 64 KB; a larger body is a 400): {"v":1, "bridge":"ops", "transport":"sse"|"websocket", "downstream":[...], "upstream":[...]}. Returns the grant (sessionId, heartbeatMs, graceMs, downstream with each topic's retain, upstream and paths). A refusal sets its status (400, 403, 429 with Retry-After, or 501) and returns {"code":..., "message":..., "denied":[...]}.

      Parameters:
      req - The REST request.
      res - The REST response.
      Returns:
      The grant, or the refusal body.

    • streamBus

      default void streamBus(@Path("sessionId") String id, RestRequest req, RestResponse res) throws IOException
      [GET /juneau-bus/stream/{sessionId}] — streams the session's frames as Server-Sent Events.

      Sends the resync, then live frames, until the client goes away or the bus closes the connection (bus closed, session expired, replaced by another connection, or too slow). Each frame is one event: bus with the frame as data and its seq, when it has one, as id.

      The stream URL contains the session capability, so an application must not log request URIs for /juneau-bus/stream/*. Juneau's own debug pipeline logs the request path of every completed call when the resolved logger is loggable at INFO or finer, and RestSession logs it at FINE when a client disconnects; keep those loggers coarser than INFO for resources that serve the bus.

      Parameters:
      id - The session's capability id.
      req - The REST request.
      res - The REST response.
      Throws:
      IOException - If the stream could not be written.