Producers: Sending & Receipts
Relevant source files
The following files were used as context for generating this wiki page:
Producers in SeaStreamer are responsible for transmitting messages to various streaming backends. They provide an asynchronous, non-blocking API that abstracts the underlying complexities of different protocols like Kafka, Redis, or local file systems. By utilizing a generic trait interface, SeaStreamer allows developers to write backend-agnostic stream processors where the producer's behavior—such as flushing and receipt handling—remains consistent regardless of the target infrastructure.
Sources: README.md:15-20, src/lib.rs:10-25, sea-streamer-socket/src/lib.rs:5-15
Producer Architecture & Lifecycle
SeaStreamer's producer architecture is built around the Producer trait, which defines the core capabilities for sending messages and managing the stream's end-of-life. Producers are typically created through a Streamer instance, which handles the initial connection and configuration.
Core Send Operations
The primary method for emitting data is send(). This operation is designed to be non-blocking; it queues the message for transmission and returns immediately, allowing the application to continue processing without waiting for network I/O. For scenarios requiring specific timing, some backends (like Redis) support send_with_ts() to specify a custom timestamp.
Sources: README.md:75-90, examples/price-feed/src/main.rs:90-95, CHANGELOG.md:25-30
Connection and Backend-Agnosticism
Using SeaProducer from the sea-streamer-socket crate allows the application to switch backends at runtime. The producer is initialized with a StreamKey and optional producer-specific configurations.
The diagram shows how the application interacts with a generic Producer interface that dispatches to specific backend implementations.
Sources: README.md:120-135, sea-streamer-socket/src/lib.rs:1-10
Receipts & Synchronization
While the send method is non-blocking, applications often need confirmation that a message has been successfully persisted. SeaStreamer handles this through "Receipts."
The Receipt Mechanism
A Receipt is a future that resolves to a MessageHeader once the backend acknowledges the message. This header contains critical metadata:
- Stream Key: The topic or stream name.
- Shard ID: The specific partition or shard.
- Sequence Number: A unique identifier for the message within the shard.
- Timestamp: The time the message was processed by the backend.
Sources: README.md:155-160, sea-streamer-file/src/producer/backend.rs:180-200
Flushing and Termination
To ensure all buffered messages are sent before a producer is closed, the flush() and end() methods are used. flush() waits for all pending receipts to complete, while end() flushes and then shuts down the producer.
Sources: README.md:92, sea-streamer-file/tests/producer.rs:65-70
File Backend Producer Implementation
The File backend (sea-streamer-file) implements a sophisticated producer system to handle concurrent writes to a single .ss file while maintaining stream integrity.
The Writers Singleton
The File backend uses a process-wide Writers manager. This component ensures that multiple FileProducer instances targeting the same file share a single FileSink. It manages a mapping of FileId to Writer tasks.
This sequence illustrates how a FileProducer communicates with the background Writer task to perform I/O and return receipts.
Sources: sea-streamer-file/src/producer/backend.rs:25-50, sea-streamer-file/src/producer/backend.rs:115-130
Stream State Tracking
The Writer maintains a StreamState for every (StreamKey, ShardId) combination. This state includes:
- The current Sequence Number.
- The latest Timestamp.
- A Running Checksum (using CRC) for data integrity.
If a producer connects to an existing file, the Writer performs a "seek-back" operation, scanning the file (aided by Beacons) to reconstruct the last known state of the stream before appending new messages.
Sources: sea-streamer-file/src/producer/backend.rs:52-65, sea-streamer-file/src/producer/backend.rs:135-175
Backend Feature Comparison
| Feature | Kafka Backend | Redis Backend | File Backend | Stdio Backend |
|---|---|---|---|---|
| Send Mode | Async wrapped sync | Pipelined XADD | Async Task / Channel | Real-time Stream |
| Receipts | Supported (via rdkafka) | Supported (auto-ack/commit) | Supported (via Writer) | Not Applicable |
| Flushing | Flush on disconnect | Background commit | Manual/Auto-flush | Synchronous Write |
| Concurrency | Non-blocking &mut | Mutex-free / Msg passing | Singleton Writer manager | Round-robin / Grouped |
Sources: README.md:155-175, sea-streamer-file/src/producer/backend.rs:20-40, sea-streamer-stdio/Cargo.toml:1-10
Usage Example: Sending & Receipts
This example demonstrates creating a producer and handling the non-blocking send with a final flush.
// Create a producer for a specific stream key
let producer: SeaProducer = streamer
.create_producer(stream.stream_key()?, Default::default())
.await?;
// Send is non-blocking
for tick in 0..100 {
let message = format!(r#""tick {tick}""#);
producer.send(message)?;
}
// flush() or end() to ensure all receipts are received
producer.end().await?;
Sources: README.md:75-95, src/lib.rs:80-100
Summary
Producers in SeaStreamer provide a unified interface for high-performance stream writing. By decoupling the send loop from the processing loop through asynchronous message passing, SeaStreamer maximizes throughput and minimizes latency. Whether using the specialized state-tracking of the File backend or the high-volume pipelining of the Redis backend, the system ensures reliable message delivery through its robust Receipt and flushing mechanisms.
Sources: README.md:190-210, sea-streamer-file/src/producer/backend.rs:215-225