Package ai.mindconnect.adminui.service

Class ActiveStreams

java.lang.Object
ai.mindconnect.adminui.service.ActiveStreams
@Component public class ActiveStreams extends Object
Server-side registry of currently-live SSE streams. Generic for any stream that wants to expose itself to the admin-UI's "is something still running?" query — chat turns, background research jobs, batch imports, etc.

Keyed by channelId — the same identifier the client sees in the Sui-Stream-Channel response header on the SSE response. The client uses it to detect re-mount and replay buffered patches; the server uses it to answer "is this stream still running?" and to cancel on demand.

Each entry holds a ActiveStreams.Handle carrying lifecycle metadata plus a caller-supplied cancel callback. The owning endpoint is responsible for registering on stream-start and deregistering on onCompletion/onError/onTimeout. Registry is intentionally process-local (in-memory ConcurrentHashMap) — multi-node deployments would need a Redis-backed implementation, but for the current single-node admin-ui this keeps the moving parts minimal.

  • Constructor Details

    • ActiveStreams

      public ActiveStreams()

  • Method Details

    • register

      public void register(ActiveStreams.Handle handle)
      Registers (or replaces) the stream with id channelId. Replace-semantics rather than reject-duplicate so a stale entry from a crashed run doesn't permanently block a re-attempt — the latest registration wins, which matches the "live reader keeps reading" model.

    • deregister

      public void deregister(String channelId)
      Drops the stream from the registry. No-op if it wasn't there.

    • cancel

      public boolean cancel(String channelId)
      Invokes the cancel callback on the named stream, if present. Returns true if a cancel was dispatched, false when the stream wasn't registered (already finished, never started, …).

      Cancellation is cooperative: the registered callback typically flips a token the streaming worker polls at safe checkpoints; the stream then ends and deregisters itself via its lifecycle hooks.

    • findHandle

      public Optional<ActiveStreams.Handle> findHandle(String channelId)

    • snapshot

      public Collection<ActiveStreams.Handle> snapshot()