Stream Handling

Relevant source files

The following files were used as context for generating this wiki page:

• src/flext_grpc/services/stream.py
• src/flext_grpc/protocols.py
• src/flext_grpc/README.md
• src/README.md
• src/flext_grpc/_utilities/grpc.py
• examples/02_advanced_usage.py

Stream Handling in the FLEXT gRPC platform encompasses the management, processing, and orchestration of various gRPC streaming patterns, including unary, server-side streaming, client-side streaming, and bidirectional streaming. The system is designed following Clean Architecture and Domain-Driven Design (DDD) principles, utilizing a dedicated service mixin and manager to handle stream lifecycle and data flow with integrated metrics and buffering.

The core logic resides in the Application Layer, specifically within the FlextGrpcStream service, which interacts with Domain Layer entities to ensure that all streaming operations are type-safe and validated against business rules.

Sources: src/README.md:23-40, src/flext_grpc/services/stream.py:1-20

Stream Architecture and Components

The stream handling architecture is split into three primary layers: the Domain Layer defining entities and types, the Application Layer providing the logic for stream orchestration, and the Infrastructure Layer facilitating interaction with the underlying gRPC runtime.

Core Components

Component	Layer	Description
FlextGrpcStream	Application	Service mixin providing the public interface for stream creation, closing, and data transmission.
GrpcStreamManager	Application	Internal manager that tracks active stream states and manages bounded message buffers.
_StreamRuntimeState	Application	Internal value object tracking a stream's instance, creation time, and message buffer.
FlextGrpcStream (Entity)	Domain	Domain entity representing a gRPC stream, containing its ID, method name, and type.
Streaming (Protocol)	Domain	Protocol definition for gRPC streaming operations like send_data and close_stream.

Sources: src/flext_grpc/README.md:29-37, src/flext_grpc/services/stream.py:22-45, src/flext_grpc/protocols.py:112-145

Stream Lifecycle Flow

The following diagram illustrates the flow from stream creation to data transmission and eventual cleanup.

Stream Lifecycle Management

Stream Creation

Stream creation involves validating the method_name and stream_type against permitted constants (e.g., unary, server_streaming, client_streaming, bidirectional). The platform utilizes a uuid4 generator for stream identification and tracks the creation timestamp for performance metrics.

def create_stream(
    self,
    method_name: str = "DefaultMethod",
    stream_type: str = "unary",
) -> p.Result[m.Grpc.GrpcStream]:
    """Create and register stream runtime state using the dedicated manager."""
    if not method_name.strip():
        return e.fail_validation("method_name", error="cannot be empty")
    if stream_type not in c.Grpc.STREAM_TYPES:
        return r[m.Grpc.GrpcStream].fail(f"Invalid stream type: {stream_type}")
    return self._stream_manager.create_stream(
        method_name=method_name,
        stream_type=stream_type,
    )

Sources: src/flext_grpc/services/stream.py:127-142, src/flext_grpc/_utilities/grpc.py:255-267

Data Transmission and Buffering

Data sent through a stream is stored in a deque buffer with a maximum size defined by c.Grpc.STREAMING_DEFAULT_BUFFER_SIZE. This ensures that high-volume streams do not consume excessive memory while allowing for asynchronous processing.

1. State Validation: The manager validates the stream state using Pydantic model_validate.
2. Payload Encapsulation: Data is wrapped in a m.Grpc.Payload which includes the data string and the current buffer size.
3. Metrics Collection: Every data operation records timestamps and stream keys for observability.

Sources: src/flext_grpc/services/stream.py:84-115, src/flext_grpc/services/stream.py:35-45

Streaming Protocols and Types

The platform defines strict protocols for streaming to ensure compatibility with different gRPC runtime implementations. The p.Grpc.Streaming protocol dictates the necessary interface for any streaming implementation.

Streaming Interface Definition

The Streaming protocol includes methods for handling various stream directions:

• handle_client_streaming: Processes a sequence of messages and returns a single response.
• handle_server_streaming: Takes a request and returns a sequence of messages.
• handle_bidirectional_streaming: Manages ongoing two-way communication.

Stream Types

The platform supports four primary gRPC operations as defined in c.Grpc.GrpcOperations:

1. Unary: Single request, single response.
2. Server Streaming: Single request, multiple responses.
3. Client Streaming: Multiple requests, single response.
4. Bidirectional: Multiple requests, multiple responses.

Sources: src/flext_grpc/services/stream.py:133-142, examples/02_advanced_usage.py:113-118

Observability and Metrics

Stream handling is integrated with FlextGrpcMetrics to provide real-time visibility into stream performance and health.

Tracked Metrics

• Stream Creation: Records the exact epoch time when a stream is initialized.
• Data Latency: Calculates average latency for messages sent through the stream.
• Error Counts: Tracks validation or connection failures specific to a stream instance.
• Buffer Status: Monitors the maxlen and current utilization of the bounded message buffer.

Sources: src/flext_grpc/services/stream.py:59-60, src/flext_grpc/services/stream.py:78-80, tests/unit/test_api.py:146-155

Usage Example: Advanced Streaming Setup

Advanced usage patterns demonstrate how to create multiple stream types through the FlextGrpc facade.

def demonstrate_streaming(self) -> None:
    """Demonstrate streaming operations through facade."""
    stream_configs = [
        ("UnaryMethod", "unary"),
        ("ServerStreamingMethod", "server_streaming"),
        ("ClientStreamingMethod", "client_streaming"),
        ("BidirectionalMethod", "bidirectional"),
    ]
    for method_name, stream_type in stream_configs:
        stream_result = self.grpc.create_stream(
            method_name=method_name,
            stream_type=stream_type,
        )
        if stream_result.success:
            stream = stream_result.value
            print(f"Created {stream_type} stream: {stream.id}")

Sources: examples/02_advanced_usage.py:112-129

Conclusion

Stream Handling in FLEXT gRPC provides a robust, type-safe framework for managing gRPC communications. By centralizing stream lifecycle management in the GrpcStreamManager and enforcing strict protocols, the system ensures reliable data flow with built-in observability and memory-efficient buffering.