Streaming & Push Transports

Audience: application developers choosing how subscribers receive events (pull vs push), and builders of LLM-style streaming workloads. Covers long-poll, SSE, WebSocket, and the two streaming-session modes.

---

Choosing a transport

Transport	Endpoint	Direction	Port	Best for
HTTP long-poll	GET /events/poll	client-driven	10105	Batch consumers, scheduled jobs, serverless; NAT-friendly
SSE	GET /events/stream	server push (one-way)	10105	Browser/mobile push, dashboards, LLM token streams
WebSocket	WS upgrade on the dedicated port	server push, bi-directional	10107 (opt-in)	Low-latency interactive clients

All three deliver identical CloudEvent payloads and share the same ACK / retry / quota machinery — the only difference is the push direction.

Long-poll

curl "http://localhost:10105/events/poll?clientId=order-svc&timeoutMs=30000"
# → [{ "deliveryId": "...", "event": { ...CloudEvent... } }, ...]

Batches up to max (default 100) buffered events; blocks up to timeoutMs. The SDK wraps this in a background loop: client.subscribe("orders", "LOAD_BALANCE", handler).

SSE

curl -N "http://localhost:10105/events/stream?clientId=order-svc" \
  -H "Accept: text/event-stream"

The runtime holds the response open and writes data: <CloudEvent JSON> frames as events arrive. Write failures nack the dispatcher immediately (the event is re-dispatched rather than lost to a dead connection). SDK: client.subscribeSse(topic, mode, handler).

WebSocket

The WS server is a separate port (-Deventmesh.ws.port=10107, disabled by default) because the upgrade handshake is a different protocol negotiation. The client must configure wsUrl explicitly — pointing it at the HTTP port fails the handshake:

CloudEventsClient wsClient = CloudEventsClient.builder()
    .runtimeUrl("http://localhost:10105")   // publish / long-poll / SSE
    .wsUrl("http://localhost:10107")        // WS push
    .clientId("ws-sub").build();
wsClient.subscribeWs("orders", "BROADCAST", event -> { ... });

LLM streaming (Mode 1 / Mode 2)

EventMesh provides two streaming patterns for LLM-style workloads — token chunks flowing back, multi-turn conversation context — built on the session layer (eventmesh-runtime/.../session/, SessionRouter).

Mode	Use case	Direction	Entry
Mode 1 — streaming call	client → agent (LLM), agent streams tokens back	request/response with push	client.streaming().openSession(...)
Mode 2 — session pub/sub	producer writes chunks; consumers read via SSE	publish/subscribe	client.subscribeSession(sessionId) / openSessionPublisher(sessionId)

Mode 1 — single streaming call

try (StreamingResponse r = client.streaming()
        .openSession(OpenSession.builder().clientId(client.clientId()).build())
        .call("Summarize this document…",)) {
    while (r.hasNext()) {
        System.out.print(r.next().text());
    }
}

Mode 2 — session pub/sub

try (SessionPublisher pub = client.openSessionPublisher(sessionId)) {
    pub.publish(chunkEvent);   // consumers on subscribeSession() receive via SSE
}
client.subscribeSession(sessionId, chunk -> render(chunk));

The v2 streaming session layer is not auto-wired into the default bootstrap: an embedder builds it via the session builders (withAgentRegistrar / withMatchmaker / withSessionRouter) before start() — the channel strategy is an explicit embedder choice. See Deployment → deployment modes.

Load balancing for sessions uses the sticky recommendation model: the runtime's LoadMeter self-reports load and /session/recommend pins a client to an instance; instances never forward each other's session traffic.

Where the code lives

Piece	Location
SSE / WS connections	eventmesh-runtime/.../push/SseConnection.java, WsConnection.java
Push pump (nack on failure)	eventmesh-runtime/.../push/ConnectionPushPump.java
Long-poll channel	eventmesh-runtime/.../push/LongPollingChannel.java
WS server	eventmesh-runtime/.../http/UniWsServer.java
Session routing	eventmesh-runtime/.../session/SessionRouter.java, Matchmaker.java
Load meter	eventmesh-runtime/.../ingress/LoadMeter.java
SDK streaming	eventmesh-sdks/.../cloudevents/stream/*
Tests	StreamingSdkE2ETest, LiteStreamCallIntegrationTest, WebSocketPushIntegrationTest, TlsIntegrationTest