Class EngineClient

java.lang.Object
ai.pipestream.connector.intake.service.EngineClient
@ApplicationScoped public class EngineClient extends Object
Builds engine intake handoff requests and sends them over the configured engine transport.

The normal path uses IntakeHandoffStreamClient's long-lived bidi stream, which avoids the per-call unary listener allocation pattern that was observed to drop requests under load. The legacy unary RPC remains available only for compatibility deployments where inline PipeDoc uploads must talk to an older engine.

  • Constructor Summary

    Constructors
    Constructor
    Description
    EngineClient()
    Default constructor for CDI proxying.

  • Method Summary

    Modifier and Type
    Method
    Description
    static ai.pipestream.engine.v1.IntakeHandoffRequest
    buildHandoffRequest(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
    Build an IntakeHandoffRequest for an inline document.
    ai.pipestream.engine.v1.IntakeHandoffRequest
    buildReferenceRequest(String docId, String sourceNodeId, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
    Build the reference-only IntakeHandoffRequest (a document_ref pointer to an already-persisted repo doc) WITHOUT sending it, so the blob path can enqueue the same request to the Redis intake ingress instead of the dropped gRPC handoff.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    handoffReferenceToEngine(String docId, String sourceNodeId, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig)
    Hands off a repository-staged document reference to engine without a crawl ID.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    handoffReferenceToEngine(String docId, String sourceNodeId, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
    Hands off a repository-staged document reference to engine.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    handoffToEngine(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig)
    Hands off an inline document through the streaming engine path without a crawl ID.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    handoffToEngine(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
    Hands off an inline document through the streaming engine path.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    handoffToEngineUnary(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
    Hands off an inline document to the engine through the legacy unary RPC.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    intakeHandoff(ai.pipestream.engine.v1.IntakeHandoffRequest request)
    Hands off one document to the engine.
    ai.pipestream.engine.v1.IntakeHandoffResponse
    intakeHandoffUnary(ai.pipestream.engine.v1.IntakeHandoffRequest request)
    Sends one already-built intake handoff request over the legacy unary engine RPC.

    Methods inherited from class Object

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

  • Constructor Details

    • EngineClient

      public EngineClient()
      Default constructor for CDI proxying.

  • Method Details

    • handoffToEngine

      public ai.pipestream.engine.v1.IntakeHandoffResponse handoffToEngine(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig)
      Hands off an inline document through the streaming engine path without a crawl ID.
      Parameters:
      pipeDoc - document to hand off
      datasourceId - datasource that owns the document
      accountId - account that owns the datasource
      ingestionConfig - Tier 1 ingestion configuration stamped by intake
      Returns:
      engine admission response

    • handoffToEngine

      public ai.pipestream.engine.v1.IntakeHandoffResponse handoffToEngine(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
      Hands off an inline document through the streaming engine path.
      Parameters:
      pipeDoc - document to hand off
      datasourceId - datasource that owns the document
      accountId - account that owns the datasource
      ingestionConfig - Tier 1 ingestion configuration stamped by intake
      crawlId - optional crawl session identifier
      Returns:
      engine admission response

    • handoffToEngineUnary

      public ai.pipestream.engine.v1.IntakeHandoffResponse handoffToEngineUnary(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
      Hands off an inline document to the engine through the legacy unary RPC.

      This is intended for compatibility deployments where inline gRPC PipeDoc uploads should bypass Redis ingress and the target engine still expects EngineV1Service/IntakeHandoff.

      Parameters:
      pipeDoc - document to hand off
      datasourceId - datasource that owns the document
      accountId - account that owns the datasource
      ingestionConfig - Tier 1 ingestion configuration stamped by intake
      crawlId - optional crawl session identifier
      Returns:
      engine admission response

    • handoffReferenceToEngine

      public ai.pipestream.engine.v1.IntakeHandoffResponse handoffReferenceToEngine(String docId, String sourceNodeId, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig)
      Hands off a repository-staged document reference to engine without a crawl ID.
      Parameters:
      docId - The Pipestream internal document identifier
      sourceNodeId - The identifier of the source node
      datasourceId - The identifier of the datasource
      accountId - The identifier of the account
      ingestionConfig - The ingestion configuration
      Returns:
      The engine's response to the handoff request

    • handoffReferenceToEngine

      public ai.pipestream.engine.v1.IntakeHandoffResponse handoffReferenceToEngine(String docId, String sourceNodeId, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
      Hands off a repository-staged document reference to engine.
      Parameters:
      docId - The Pipestream internal document identifier
      sourceNodeId - The identifier of the source node
      datasourceId - The identifier of the datasource
      accountId - The identifier of the account
      ingestionConfig - The ingestion configuration
      crawlId - The identifier of the crawl session (optional)
      Returns:
      The engine's response to the handoff request

    • intakeHandoff

      public ai.pipestream.engine.v1.IntakeHandoffResponse intakeHandoff(ai.pipestream.engine.v1.IntakeHandoffRequest request)
      Hands off one document to the engine.

      Delegates to IntakeHandoffStreamClient, which holds a single long-lived bidi stream open to the engine and multiplexes per-doc requests over it. Replaced the prior per-call unary stub.intakeHandoff(req, observer) pattern: that path went through Quarkus's BlockingServerInterceptor.VirtualReplayListener on every request and was observed to drop ~5/1000 docs under load with INTERNAL: Half-closed without a request. The bidi RPC avoids the per-call buffer-allocation pattern that triggers the drops.

      Method signature unchanged from the unary version — every existing caller (handoffToEngine, handoffReferenceToEngine, blob upload, IntakeRepoEventConsumer) automatically becomes bidi.

      Parameters:
      request - The handoff request to send to the engine
      Returns:
      The engine's response to the handoff request

    • intakeHandoffUnary

      public ai.pipestream.engine.v1.IntakeHandoffResponse intakeHandoffUnary(ai.pipestream.engine.v1.IntakeHandoffRequest request)
      Sends one already-built intake handoff request over the legacy unary engine RPC.

    • buildHandoffRequest

      public static ai.pipestream.engine.v1.IntakeHandoffRequest buildHandoffRequest(ai.pipestream.data.v1.PipeDoc pipeDoc, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
      Build an IntakeHandoffRequest for an inline document.
      Parameters:
      pipeDoc - The document to include in the request
      datasourceId - The identifier of the datasource
      accountId - The identifier of the account
      ingestionConfig - The ingestion configuration
      crawlId - The identifier of the crawl session (optional)
      Returns:
      A constructed IntakeHandoffRequest

    • buildReferenceRequest

      public ai.pipestream.engine.v1.IntakeHandoffRequest buildReferenceRequest(String docId, String sourceNodeId, String datasourceId, String accountId, ai.pipestream.data.v1.IngestionConfig ingestionConfig, String crawlId)
      Build the reference-only IntakeHandoffRequest (a document_ref pointer to an already-persisted repo doc) WITHOUT sending it, so the blob path can enqueue the same request to the Redis intake ingress instead of the dropped gRPC handoff.