streaming

func ProcessBidiStream ¶

func ProcessBidiStream[InT, OutT any](
	ctx context.Context,
	stream BidiStream[InT, OutT],
	inputCh <-chan *InT,
) (StreamResult[OutT], error)

ProcessBidiStream handles concurrent bidirectional streaming.

Pattern: Sender || Receiver (parallel goroutines)

This processor implements true bidirectional streaming with concurrent send and receive operations. It spawns two independent goroutines:

• Sender: Continuously sends inputs from the input channel
• Receiver: Continuously receives outputs and sends them to the output channel

This pattern maximizes throughput by:

• Eliminating round-trip latency between requests
• Allowing the server to batch/buffer/pipeline operations
• Fully utilizing network bandwidth
• Enabling concurrent processing on both client and server

This is useful when:

• High performance and throughput are needed
• Processing large batches of data
• Server can process requests in parallel or batches
• Responses can arrive in any order or timing
• Network latency is significant

Returns:

• result: StreamResult containing result, error, and done channels
• error: Immediate error if validation fails

The caller should:

1. Range over result channels to process outputs and errors
2. Check if the processing is done
3. Use context cancellation to stop processing early

func ProcessClientStream ¶

func ProcessClientStream[InT, OutT any](
	ctx context.Context,
	stream ClientStream[InT, OutT],
	inputCh <-chan *InT,
) (StreamResult[OutT], error)

ProcessClientStream handles client streaming pattern (many inputs → one output).

Pattern: Send → Send → Send → CloseAndRecv()

This processor is ideal for operations where multiple requests are sent to the server, and a single final response is received after all requests have been processed.

The processor:

• Sends all inputs from the channel to the stream
• Closes the send side when input channel closes
• Receives the final response via CloseAndRecv()

Returns:

• result: StreamResult containing result, error, and done channels
• error: Immediate error if validation fails

The caller should:

1. Range over result channels to process outputs and errors
2. Check if the processing is done
3. Use context cancellation to stop processing early

func ProcessServerStream ¶

func ProcessServerStream[OutT any](
	ctx context.Context,
	stream ServerStream[OutT],
) (StreamResult[OutT], error)

ProcessServerStream handles server streaming pattern (one input → many outputs).

Pattern: Send Request → Recv() → Recv() → Recv() → EOF

This processor is ideal for operations where a single request triggers multiple responses from the server, such as:

• Streaming search results
• Listing resources
• Tailing logs
• Real-time event streams

The processor:

• Continuously receives outputs from the stream
• Sends each output to the result channel
• Handles EOF gracefully to signal completion
• Propagates errors to the error channel

Returns:

• result: StreamResult containing result, error, and done channels
• error: Immediate error if validation fails

The caller should:

1. Range over result channels to process outputs and errors
2. Monitor the DoneCh to know when streaming is complete
3. Use context cancellation to stop processing early

Example usage:

stream, err := client.Listen(ctx, req)
if err != nil {
    return err
}

result, err := streaming.ProcessServerStream(ctx, stream)
if err != nil {
    return err
}

for {
    select {
    case resp := <-result.ResCh():
        // Process response
    case err := <-result.ErrCh():
        // Handle error
        return err
    case <-result.DoneCh():
        // All responses received
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}