Class SingleDocDispatcher

java.lang.Object

ai.pipestream.module.pipelineprobe.pipelinecrawl.SingleDocDispatcher

@ApplicationScoped
public class SingleDocDispatcher
extends Object

Drives the SINGLE_DOC source for a RunPipelineCrawl run.

Two intake modes, selected by SingleDocSourceConfig.getIntakeMode():

• HTTP (default): streams the sample to connector-intake's HTTP raw upload (POST /uploads/raw) — the S3-style path. The body flows connector-intake → repo → S3 chunk-by-chunk and lands in the account's :intake drive, never buffered whole.
• gRPC: builds a PipeDoc with the sample bytes in its BlobBag and sends it via the unary UploadPipeDoc — the in-memory path for small docs. connector-intake enqueues it to the Redis intake ingress (same as HTTP), so the engine drains it to :intake and the parser hydrates the blob. A client could equally put its own model in structured_data (Any) instead of a blob.

In both modes the handoff rides the Redis intake ingress and the engine drains it like every other intake.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	SingleDocDispatcher.SingleDocDispatch	Result of uploadAndDispatch(String, String, String, SingleDocSourceConfig, String): the doc_id the upload produced.

Constructor Summary

Constructors

Constructor	Description
SingleDocDispatcher()	Creates the single-doc dispatcher.

Method Summary

Modifier and Type	Method	Description
SingleDocDispatcher.SingleDocDispatch	uploadAndDispatch(String accountId, String datasourceId, String apiKey, ai.pipestream.testing.harness.v1.SingleDocSourceConfig cfg, String crawlId)	Upload a single bundled sample to connector-intake and return the resulting doc_id.

Methods inherited from class Object

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

Constructor Details

SingleDocDispatcher

public SingleDocDispatcher()

Creates the single-doc dispatcher. Managed by CDI.

Method Details

uploadAndDispatch

public SingleDocDispatcher.SingleDocDispatch uploadAndDispatch(String accountId,
 String datasourceId,
 String apiKey,
 ai.pipestream.testing.harness.v1.SingleDocSourceConfig cfg,
 String crawlId)

Upload a single bundled sample to connector-intake and return the resulting doc_id. The intake mode (HTTP raw stream vs gRPC unary UploadPipeDoc with a BlobBag) is chosen by cfg.intake_mode; E2EIntakeMode.E2E_INTAKE_MODE_GRPC selects gRPC, anything else (including UNSPECIFIED) uses the HTTP path.

Parameters:

accountId - the run's account id (stamped on the source_doc_id)

datasourceId - the datasource minted by DatasourceManager

apiKey - api key for that datasource

cfg - the single_doc branch of the source config

crawlId - the run's crawl id, stamped on the upload so the doc's StepExecutionRecords carry it (live observation panels)

Returns:

the SingleDocDispatcher.SingleDocDispatch carrying the resulting doc_id

Throws:

IllegalStateException - if the upload fails