Package ai.pipeline.connector.intake.service

Class ConnectorIntakeServiceImpl

java.lang.Object
ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase
ai.pipeline.connector.intake.service.ConnectorIntakeServiceImpl
All Implemented Interfaces:
io.grpc.BindableService
public class ConnectorIntakeServiceImpl extends ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase
gRPC service implementation for Connector-Intake.

Handles document streaming, crawl session management, and heartbeat monitoring.

Proto: grpc/grpc-stubs/src/main/proto/module/connectors/connector_intake_service.proto

  • Constructor Summary

    Constructors
    Constructor
    Description
    ConnectorIntakeServiceImpl()
     

  • Method Summary

    Modifier and Type
    Method
    Description
    io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.EndCrawlSessionResponse>
    endCrawlSession(ai.pipestream.connector.intake.EndCrawlSessionRequest request)
    End a crawl session and persist final statistics.
    io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.HeartbeatResponse>
    heartbeat(ai.pipestream.connector.intake.HeartbeatRequest request)
    Heartbeat endpoint for long-running crawls.
    io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.StartCrawlSessionResponse>
    startCrawlSession(ai.pipestream.connector.intake.StartCrawlSessionRequest request)
    Start a new crawl session using the unary RPC endpoint.
    io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.DocumentIntakeResponse>
    streamDocuments(io.smallrye.mutiny.Multi<ai.pipestream.connector.intake.DocumentIntakeRequest> requests)
    Stream processing entry point for the Connector Intake gRPC service.

    Methods inherited from class ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase

    bindService, withCompression

    Methods inherited from class java.lang.Object

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

  • Constructor Details

    • ConnectorIntakeServiceImpl

      public ConnectorIntakeServiceImpl()

  • Method Details

    • streamDocuments

      public io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.DocumentIntakeResponse> streamDocuments(io.smallrye.mutiny.Multi<ai.pipestream.connector.intake.DocumentIntakeRequest> requests)
      Stream processing entry point for the Connector Intake gRPC service.

      CLIENT-SIDE STREAMING: Client sends all chunks, server returns ONE final response.

      Consumes a stream of DocumentIntakeRequest messages and returns a single DocumentIntakeResponse after all chunks are processed. Requests can be one of:

      • SessionStart – must be sent first to authenticate and initialize session.
      • Document – document payload and metadata; may be raw or chunked.
      Reactive semantics:
      • All incoming requests are collected and processed in parallel.
      • Returns ONE final response with batch results after stream completes.
      • No intermediate responses - client can fire all chunks without waiting.
      Side effects:
      • On SessionStart: validates the connector and creates a persisted crawl session.
      • On Document: writes content to the repository-service in parallel.
      Overrides:
      streamDocuments in class ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase
      Parameters:
      requests - stream of intake requests for a logical crawl session
      Returns:
      single final response with batch processing results

    • startCrawlSession

      public io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.StartCrawlSessionResponse> startCrawlSession(ai.pipestream.connector.intake.StartCrawlSessionRequest request)
      Start a new crawl session using the unary RPC endpoint.

      Validates the connector credentials and creates a persisted crawl session with optional tracking and orphan-deletion behavior as requested by the client. Metadata is serialized and stored with the session for downstream processing. Reactive semantics:

      • Returns a Uni that completes on the default worker pool due to persistence work.
      • Failure is mapped to a response with success=false and an explanatory message.
      Side effects: creates or reuses a session row in the database.
      Overrides:
      startCrawlSession in class ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase
      Parameters:
      request - request containing connector ID, API key, crawl metadata and options
      Returns:
      a Uni emitting StartCrawlSessionResponse with the created session ID on success

    • endCrawlSession

      public io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.EndCrawlSessionResponse> endCrawlSession(ai.pipestream.connector.intake.EndCrawlSessionRequest request)
      End a crawl session and persist final statistics.

      Looks up the session, applies summary statistics if provided, and transitions the session to a completed state. If the session does not exist, a response with success=false is returned. Reactive semantics:

      • Runs on the default worker pool due to database access.
      • Errors are mapped to a success=false response.
      Side effects: updates the session row and marks completion.
      Overrides:
      endCrawlSession in class ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase
      Parameters:
      request - request containing the session ID and optional CrawlSummary
      Returns:
      a Uni emitting EndCrawlSessionResponse

    • heartbeat

      public io.smallrye.mutiny.Uni<ai.pipestream.connector.intake.HeartbeatResponse> heartbeat(ai.pipestream.connector.intake.HeartbeatRequest request)
      Heartbeat endpoint for long-running crawls.

      Validates that the session exists, updates its heartbeat timestamp, and returns a control command to the client. For now the command is always COMMAND_CONTINUE when the session is valid; otherwise COMMAND_STOP is returned. Reactive semantics: executes on the default worker pool due to database access. Side effects: updates the session's last heartbeat time.

      Overrides:
      heartbeat in class ai.pipestream.connector.intake.MutinyConnectorIntakeServiceGrpc.ConnectorIntakeServiceImplBase
      Parameters:
      request - heartbeat request containing sessionId and crawlId
      Returns:
      a Uni that emits the heartbeat response indicating session validity and next command