Class PipelineCrawlerResource

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.PipelineCrawlerResource
@Path("/test-sidecar/v1/pipeline-crawl") public class PipelineCrawlerResource extends Object
REST/SSE bridge from the FE's Pipeline tab to the gRPC PipelineCrawlerImpl. Uses JAX-RS native server-sent events (Sse + SseEventSink) so we stay on plain Java the whole way through — no Mutiny anywhere in the pipeline-crawl path.

The endpoint is marked RunOnVirtualThread so the JAX-RS runtime invokes the method on a VT. The run itself executes on a RunRegistry-owned virtual thread — this resource only starts it and subscribes the SSE sink as one of its watchers (journal replay + live tail). A client disconnect detaches the watcher; the run is unaffected and remains re-attachable via GET runs/{runId}/events until ~30 minutes after it terminates.

Why not Multi<String> like the legacy resources: that pattern pulls Mutiny back into the new code path. JAX-RS native SSE is the platform-native way to stream events from a blocking endpoint.

  • Constructor Details

    • PipelineCrawlerResource

      public PipelineCrawlerResource()
      Creates the pipeline-crawl REST/SSE resource. Collaborators (the gRPC crawler and the run registry) are injected; instances are managed by the JAX-RS / CDI runtime.

  • Method Details

    • runPipelineCrawl

      @POST @Consumes("application/json") @Produces("text/event-stream") public void runPipelineCrawl(PipelineCrawlerResource.PipelineCrawlPayload payload, @Context jakarta.ws.rs.sse.Sse sse, @Context jakarta.ws.rs.sse.SseEventSink sink)
      Start a pipeline-crawl run and stream its events over SSE. Translates the FE payload to the proto request, hands it to the RunRegistry to execute on a registry-owned virtual thread, then subscribes this SSE response as a watcher (journal replay + live tail). A client disconnect detaches the watcher only; the run continues and stays re-attachable via GET runs/{runId}/events. An invalid request shape or a failed start is reported as a single SSE error event; the sink is closed in the finally block.
      Parameters:
      payload - the FE-facing crawl request payload
      sse - the JAX-RS SSE factory used to build outbound events
      sink - the SSE event sink this response writes to

    • listRuns

      @GET @Path("/runs") @Produces("application/json") public Map<String,Object> listRuns()
      List known runs (live + recently terminal). Each row carries the run_id used for re-attach and cancel.
      Returns:
      a JSON-shaped map with key "runs" holding one row per known run (run_id, source, state, timestamps, journal stats and last message)

    • streamRunEvents

      @GET @Path("/runs/{runId}/events") @Produces("text/event-stream") public void streamRunEvents(@PathParam("runId") String runId, @Context jakarta.ws.rs.sse.Sse sse, @Context jakarta.ws.rs.sse.SseEventSink sink)
      Re-attach to a run: full journal replay, then live tail. Works for live runs (watch resumes) and recently-terminal ones (replay then close). Unknown/evicted run ids answer with a single error event.
      Parameters:
      runId - the id of the run to re-attach to
      sse - the JAX-RS SSE factory used to build outbound events
      sink - the SSE event sink this response writes to

    • cancelRun

      @POST @Path("/runs/{runId}/cancel") @Produces("application/json") public Map<String,Object> cancelRun(@PathParam("runId") String runId)
      Cancel a live run. The run thread is interrupted; the orchestration maps that to a FAILED summary + ninja cleanup, so cancellation is clean by construction.
      Parameters:
      runId - the id of the run to cancel
      Returns:
      a JSON-shaped map with the run_id, a cancelled flag and a reason (unknown run_id, interrupt sent, or already terminal)