Package org.codelibs.fess.api.v2.handlers

Class ChatStreamHandler

java.lang.Object
org.codelibs.fess.api.v2.handlers.ChatStreamHandler
public class ChatStreamHandler extends Object
Handles POST /api/v2/chat/stream — Server-Sent Events streaming RAG chat.

Envelope exception: SSE wire format is incompatible with the {"response": {...}} JSON envelope used by every other v2 endpoint except /api/v2/documents/all (NDJSON). Plan 4 deliberately keeps the same SSE framing as v1 (event: + data: <json>\n\n) so the static theme JS can share a single SSE parser across v1 and v2. The exception is documented in §Risks (Risk 2).

All SSE event data keys use snake_case:

  • event: phase — {phase, status: start|complete, message?, keywords?, hit_count?}
  • event: chunk — {content}
  • event: sources — {sources: [{rank, title, url, doc_id, snippet, url_link, go_url}]}
  • event: done — {session_id, html_content?}
  • event: retry — {phase, operation, attempt, max_attempts, sleep_ms, cause?}
  • event: waiting — {phase, reason, elapsed_ms, timeout_ms}
  • event: fallback — {phase, reason, original_query?, new_query?}
  • event: warning — {phase, code, detail?}
  • event: error — {phase?, message, error_code}

Error reporting before the LLM is invoked (method check, feature gate, body parse, rate limit) uses V2EnvelopeWriter.writeError(jakarta.servlet.http.HttpServletResponse, org.codelibs.fess.api.v2.V2ErrorCode, java.lang.String) so the HTTP status and Content-Type: application/json are correct. Only after all gates pass are SSE headers set; subsequent LLM-level errors are reported via event: error SSE events, consistent with v1 behaviour the static theme JS parser depends on.

  • Constructor Details

    • ChatStreamHandler

      public ChatStreamHandler()
      Default constructor used by the DI container. The handler holds no per-request state and is safe to share across concurrent requests. init() must be called (by the DI container via PostConstruct) before any request is processed.

  • Method Details

    • init

      @PostConstruct public void init()
      Initializes the shared keep-alive scheduler pool. Called by Lasta Di after the component is bound (PostConstruct).

      The pool uses daemon threads named fess-sse-keepalive-N (1-based) so they are visible in thread dumps and never prevent JVM shutdown.

    • destroy

      @PreDestroy public void destroy()
      Shuts down the shared keep-alive pool on container shutdown (PreDestroy).

      Waits up to 1 second for in-flight tasks to complete; if they have not finished by then, ExecutorService.shutdownNow() is called to interrupt waiting threads. Individual per-request futures are cancelled via Future.cancel(boolean) before this point, so only the pool lifecycle itself needs termination here.

    • handle

      public void handle(jakarta.servlet.http.HttpServletRequest req, jakarta.servlet.http.HttpServletResponse res) throws IOException
      Processes one POST /api/v2/chat/stream request.

      All pre-stream validation (method, feature gate, body parse, rate limit) runs before any SSE headers are committed so that failures can be reported as a normal JSON error envelope. Once every gate passes the response is switched to text/event-stream and the LLM is invoked; subsequent failures surface via an event: error SSE event rather than an HTTP error code.

      Parameters:
      req - the incoming HTTP request
      res - the HTTP response to write to
      Throws:
      IOException - if writing the envelope or reading the body fails

    • getUserId

      protected String getUserId(jakarta.servlet.http.HttpServletRequest req)
      Resolves the effective chat user id. Exposed as a seam so unit tests can override the user identity directly. SystemHelper/UserInfoHelper are smart-deploy components, so stubbing them via ComponentUtil.register is not reliable once the shared test container has resolved the real ones; overriding this method avoids that fragility.
      Parameters:
      req - the incoming HTTP request
      Returns:
      the user identifier (never null)

    • getRateLimitKey

      protected String getRateLimitKey(jakarta.servlet.http.HttpServletRequest req)
      Resolves the chat rate-limit key (server-validated username for authenticated callers, else the proxy-aware client IP). Exposed as a seam so unit tests can pin the throttle key deterministically; the underlying SystemHelper/RateLimitHelper are smart-deploy components that are unreliable to stub via ComponentUtil.register once the shared test container has resolved the real ones (same rationale as getUserId(jakarta.servlet.http.HttpServletRequest)).
      Parameters:
      req - the incoming HTTP request
      Returns:
      the rate-limit key (never null/blank)

    • resolveKeepaliveIntervalMs

      protected long resolveKeepaliveIntervalMs(org.codelibs.fess.mylasta.direction.FessConfig fessConfig)
      Resolves the keep-alive interval from configuration with a safe default. Returns the configured value when valid, or 15000 ms when the config key is missing, malformed, or negative. A value of 0 disables the pinger.
      Parameters:
      fessConfig - active configuration
      Returns:
      interval in milliseconds; 0 disables the pinger

    • startKeepalivePinger

      protected ScheduledFuture<?> startKeepalivePinger(PrintWriter writer, Object writeLock, long intervalMs)
      Schedules a periodic keep-alive ping task on the shared sharedKeepalivePool. Emits ": keepalive\n\n" (SSE comment line — ignored by EventSource clients) on a fixed delay. Returns null when intervalMs is <=0 so the pinger is fully disabled.

      All emissions are guarded by writeLock which is also held by the request thread while writing event/data pairs, so the pinger can never interleave inside an event frame.

      The returned ScheduledFuture must be cancelled (not the pool) when the request completes — call stopKeepalivePinger(ScheduledFuture).

      Parameters:
      writer - servlet writer
      writeLock - shared monitor serializing writes to writer
      intervalMs - ping interval; <=0 disables pinging
      Returns:
      a future representing the scheduled ping task, or null when disabled

    • stopKeepalivePinger

      protected void stopKeepalivePinger(ScheduledFuture<?> future)
      Cancels a keep-alive ping future previously returned by startKeepalivePinger(PrintWriter, Object, long). The shared pool is NOT shut down — only this request's scheduled task is cancelled. Safe to call with null.
      Parameters:
      future - the ping future to cancel; may be null

    • setSseHeaders

      protected void setSseHeaders(jakarta.servlet.http.HttpServletResponse res)
      Same SSE headers as v1. Delegates to SseResponseHelper.applySseHeaders(jakarta.servlet.http.HttpServletResponse) so the baseline (including the critical X-Accel-Buffering: no header that disables nginx response buffering) stays in sync across v1 and v2.
      Parameters:
      res - the HTTP response to set headers on

    • newPhaseCallback

      protected ChatPhaseCallback newPhaseCallback(PrintWriter writer)
      Builds a ChatPhaseCallback that emits SSE events with snake_case keys.
      Parameters:
      writer - the per-request servlet writer to emit events to
      Returns:
      a callback bound to the writer

    • newPhaseCallback

      protected ChatPhaseCallback newPhaseCallback(PrintWriter writer, boolean[] errorEmittedHolder)
      Backward-compatible overload that synthesises a private write lock. Prefer newPhaseCallback(PrintWriter, boolean[], Object) from the request thread so the lock is shared with the keep-alive pinger.
      Parameters:
      writer - the per-request servlet writer to emit events to
      errorEmittedHolder - single-element boolean array; element 0 is set to true after the first onError emission
      Returns:
      a callback bound to the writer

    • newPhaseCallback

      protected ChatPhaseCallback newPhaseCallback(PrintWriter writer, boolean[] errorEmittedHolder, Object writeLock)
      Builds a ChatPhaseCallback that emits SSE events with snake_case keys. The supplied errorEmittedHolder flag is set whenever onError is invoked so the surrounding handler can avoid double-emitting an error event from its outer catch. All emissions are serialized on writeLock so they cannot interleave with concurrent keep-alive pings.
      Parameters:
      writer - the per-request servlet writer to emit events to
      errorEmittedHolder - single-element boolean array; element 0 is set to true after the first onError emission
      writeLock - shared monitor serializing writes to writer
      Returns:
      a callback bound to the writer

    • emitSafely

      protected void emitSafely(PrintWriter writer, String event, Map<String,Object> data)
      Wraps sendSseEvent(java.io.PrintWriter, java.lang.String, java.util.Map<java.lang.String, java.lang.Object>) so a failure inside the phase callback (typically an IOException from a disconnected client) does not propagate and abort the surrounding streamChatEnhanced call.
      Parameters:
      writer - servlet writer
      event - SSE event name
      data - event payload to serialize as JSON

    • emitSafely

      protected void emitSafely(PrintWriter writer, String event, Map<String,Object> data, Object writeLock)
      Lock-aware variant of emitSafely(PrintWriter, String, Map) that serializes the write on a shared monitor — used so phase events cannot interleave with concurrent keep-alive pings inside an event/data frame.
      Parameters:
      writer - servlet writer
      event - SSE event name
      data - event payload to serialize as JSON
      writeLock - shared monitor serializing writes to writer

    • putIfNotNull

      protected void putIfNotNull(Map<String,Object> data, String key, Object value)
      Helper to add a key/value pair to a map only when the value is non-null.
      Parameters:
      data - target map
      key - key
      value - value to add when non-null

    • sendSseEvent

      protected void sendSseEvent(PrintWriter writer, String event, Map<String,Object> data)
      Writes one SSE event using v1's wire format: event: <name>\ndata: <json>\n\n.
      Parameters:
      writer - servlet writer
      event - SSE event name
      data - event payload to serialize as JSON