Package ai.pipeline.connector.intake.service

Class StreamingChunkProcessor

java.lang.Object

ai.pipeline.connector.intake.service.StreamingChunkProcessor

@ApplicationScoped
public class StreamingChunkProcessor
extends Object

Processor for handling streaming chunks from large file uploads.

This processor forwards chunks to repo-service's NodeUploadService which handles streaming to S3. SHA256 is calculated incrementally for verification.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	StreamingChunkProcessor.ProcessingResult	Result of processing a chunk.
static class	StreamingChunkProcessor.UploadResult	Upload result containing metadata about the uploaded file.

Constructor Summary

Constructors

Constructor	Description
StreamingChunkProcessor()

Method Summary

Modifier and Type	Method	Description
void	cleanupStaleSessions()	Clean up stale streaming sessions.
io.smallrye.mutiny.Uni<StreamingChunkProcessor.ProcessingResult>	processChunk(ai.pipestream.connector.intake.StreamingChunk chunk, String bucketName, String connectorId, String fileName, String path, Map<String,String> metadata)	Process a streaming chunk for a large file using header/raw/footer protocol.

Methods inherited from class java.lang.Object

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

Constructor Details

StreamingChunkProcessor

public StreamingChunkProcessor()

Method Details

processChunk

public io.smallrye.mutiny.Uni<StreamingChunkProcessor.ProcessingResult> processChunk(ai.pipestream.connector.intake.StreamingChunk chunk,
 String bucketName,
 String connectorId,
 String fileName,
 String path,
 Map<String,String> metadata)

Process a streaming chunk for a large file using header/raw/footer protocol.

This method handles three types of chunks: 1. Header chunk: Blob with storage_ref and metadata 2. Raw data chunks: Pure file content streamed to S3 3. Footer chunk: Final metadata and completion status

Parameters:

chunk - The streaming chunk data

bucketName - S3 bucket name (used as drive name)

connectorId - Connector identifier

fileName - Original file name

path - File path

metadata - File metadata

Returns:

ProcessingResult indicating completion status and upload metadata

cleanupStaleSessions

public void cleanupStaleSessions()

Clean up stale streaming sessions.