Class IntakeDocStream

java.lang.Object

ai.pipestream.module.pipelineprobe.moduletest.engineperf.IntakeDocStream

All Implemented Interfaces:

AutoCloseable

public final class IntakeDocStream
extends Object
implements AutoCloseable

One open bidi stream to connector-intake's uploadPipeDocStream — the SAME protocol the JDBC connector speaks (IntakeCrawlStream is the reference implementation): StreamContext handshake first (datasource + api key + crawl_id), then one PipeDocItem per doc with per-doc acks; backpressure is HTTP/2 native.

This is the "sidecar as its own connector" transport for the engine-path perf test: documents enter through the REAL front door and the engine does everything it does in production (claim-check staging, fan-out to the subscribed graph, work dispensing to the module's own worker loop).

Not thread-safe on the upload side — one session, one publisher thread.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	IntakeDocStream.DocAck	Per-doc intake outcome.

Constructor Summary

Constructors

Constructor	Description
IntakeDocStream(ai.pipestream.connector.intake.v1.ConnectorIntakeServiceGrpc.ConnectorIntakeServiceStub stub, String datasourceId, String apiKey, String crawlId, Duration contextAckTimeout, Duration closeTimeout)	Opens the stream, sends the context handshake, and blocks (up to contextAckTimeout) for the server's accept before returning.

Method Summary

Modifier and Type	Method	Description
void	close()	Half-close and wait (bounded) for the server's terminal completion.
CompletableFuture<IntakeDocStream.DocAck>	upload(ai.pipestream.data.v1.PipeDoc doc, String sourceDocId)	Push one doc.

Methods inherited from class Object

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

Constructor Details

IntakeDocStream

public IntakeDocStream(ai.pipestream.connector.intake.v1.ConnectorIntakeServiceGrpc.ConnectorIntakeServiceStub stub,
 String datasourceId,
 String apiKey,
 String crawlId,
 Duration contextAckTimeout,
 Duration closeTimeout)

Opens the stream, sends the context handshake, and blocks (up to contextAckTimeout) for the server's accept before returning.

Parameters:

stub - the connector-intake async stub to open the upload stream on

datasourceId - the datasource the stream uploads under (sent in the handshake)

apiKey - the api key authenticating the call via x-api-key metadata (never carried inside the StreamContext body)

crawlId - the crawl id sent in the context handshake

contextAckTimeout - how long to wait for the server to accept the handshake

closeTimeout - how long close() waits for terminal completion

Throws:

IllegalStateException - if the context wait is interrupted, times out, or the server rejects the handshake

Method Details

upload

public CompletableFuture<IntakeDocStream.DocAck> upload(ai.pipestream.data.v1.PipeDoc doc,
 String sourceDocId)

Push one doc. Parks the caller only when HTTP/2 flow control says the wire is full (intake backpressure — the production behavior).

Parameters:

doc - the document to upload

sourceDocId - the source doc id correlating this upload to its ack

Returns:

a future completing with the per-doc IntakeDocStream.DocAck; already completed with a failed, retryable ack if the stream is closed

close

public void close()

Half-close and wait (bounded) for the server's terminal completion.

Specified by:

close in interface AutoCloseable