Package ai.mindconnect.adminui.service

Class StreamBus

java.lang.Object

ai.mindconnect.adminui.service.StreamBus

public final class StreamBus
extends Object

Per-channel multiplexing fan-out used by ActiveStreams. One instance is created per live SSE stream and lives on the channel's ActiveStreams.Handle. Producers call publish(java.lang.String, java.lang.String) to push an event; every currently-attached SseEmitter sees a copy.

Each event gets a monotonically-increasing sequence number, and the last BUFFER_SIZE events are kept in a ring buffer. A client that reconnects with a lastSeq parameter via attach(org.springframework.web.servlet.mvc.method.annotation.SseEmitter, long) receives every event in the buffer with seq > lastSeq before being subscribed to the live feed — closing the small race window between a disconnect and a reconnect (typically a few seconds on F5).

Thread-safety:

• publish is synchronized so seq assignment and the buffer/subscriber writes are atomic.
• The subscriber list is a CopyOnWriteArrayList — fine for a low number of concurrent subscribers (1-3 in practice; one tab streaming, maybe one observer tab).
• attach(org.springframework.web.servlet.mvc.method.annotation.SseEmitter, long) synchronizes against publish so a freshly attached subscriber can't miss an event that's in flight while the buffer-replay loop runs.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	StreamBus.Event	One past event held in the ring buffer.

Field Summary

Fields

Modifier and Type	Field	Description
static final int	BUFFER_SIZE	Ring-buffer depth.

Constructor Summary

Constructors

Constructor	Description
StreamBus()

Method Summary

Modifier and Type	Method	Description
void	attach(org.springframework.web.servlet.mvc.method.annotation.SseEmitter emitter, long lastSeq)	Subscribes emitter to the live feed.
void	closeAll()	Completes every attached subscriber and clears the list.
void	detach(org.springframework.web.servlet.mvc.method.annotation.SseEmitter emitter)	Drops an emitter from the subscriber list.
long	lastSeq()	Last published seq, for clients that need to track "where am I".
void	publish(String name, String data)	Records the event in the ring buffer and broadcasts it to every currently attached subscriber.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

BUFFER_SIZE

public static final int BUFFER_SIZE

Ring-buffer depth. Sized for a chat-turn's worth of token-by-token patches plus task-card state changes — typical turn produces well under 200 events. Bumping this is cheap (a few KB per stream).

See Also:

• Constant Field Values

Constructor Details

StreamBus

public StreamBus()

Method Details

publish

public void publish(String name,
 String data)

Records the event in the ring buffer and broadcasts it to every currently attached subscriber. Failures on a single subscriber drop that subscriber from the list — its emitter is already broken at that point, no benefit in retrying.

attach

public void attach(org.springframework.web.servlet.mvc.method.annotation.SseEmitter emitter,
 long lastSeq)

Subscribes emitter to the live feed. If lastSeq is non-negative, every buffered event with seq > lastSeq is replayed synchronously before the emitter is added to the subscriber list — the holding lock guarantees no live event interleaves.

Pass -1 as lastSeq to replay everything in the buffer (initial attach from a reconnecting client that has no seq yet). Pass Long.MAX_VALUE to skip replay entirely (the stream's own producer thread doesn't need replay).

detach

public void detach(org.springframework.web.servlet.mvc.method.annotation.SseEmitter emitter)

Drops an emitter from the subscriber list. No-op if not present.

lastSeq

public long lastSeq()

Last published seq, for clients that need to track "where am I".

closeAll

public void closeAll()

Completes every attached subscriber and clears the list. Called by the channel's owner after the producer is fully done (after the done event has been published) so reconnect-tabs see a clean stream end rather than hanging on an open SSE forever.