Package com.axelor.ai.chat.stream

Class ActiveStream

java.lang.Object
com.axelor.ai.chat.stream.ActiveStream
All Implemented Interfaces:
AutoCloseable
public class ActiveStream extends Object implements AutoCloseable
Generation-side handle for one chat run. The agentic loop appends serialized SSE frames; frames are flushed to the registry's chunk log and announced on the topic. Under a distributed cache provider frames are micro-batched (~50L ms) to keep per-token network roundtrips bounded; under caffeine every append flushes inline.

The cancel signal arrives via the registry topic: isCancelRequested() flips and the onCancel(java.lang.Runnable) callback runs (immediately, if the signal already arrived).

  • Method Details

    • append

      public void append(String frames)
      Appends serialized SSE frames. Throws StreamOverflowException past the size cap.

    • markTurn

      public void markTurn(String messageId)
      Records the assistant message this run generates, from its opening frame. First frame wins: a resuming client uses it to tell which of its persisted turns this run's replay rebuilds.

    • terminating

      public void terminating()
      Exempts subsequent appends from the size cap (error/abort/terminator frames).

    • done

      public void done()
      Flushes, marks the run done, releases the claim. Idempotent. Always stops the flusher and removes the listener even if a cache/topic call throws mid-way, so a failure at completion can never leak the virtual thread or the topic listener (or leave closed false, which would keep the flusher spinning and the claim blocked).

    • onCancel

      public void onCancel(Runnable callback)
      Registers the cancel callback; runs it immediately if cancel was already signalled.

    • isCancelRequested

      public boolean isCancelRequested()

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable