Backend-Agnostic Socket API

Relevant source files

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

• sea-streamer-socket/src/lib.rs
• sea-streamer-socket/src/streamer.rs
• sea-streamer-socket/src/backend.rs
• README.md
• src/lib.rs
• sea-streamer-socket/Cargo.toml

The Backend-Agnostic Socket API, implemented primarily in the sea-streamer-socket crate, provides a concrete-type API that allows Rust applications to interact with various streaming backends at runtime without recompilation. Similar to how SeaORM abstracts different databases, this module abstracts streaming servers like Kafka, Redis, Stdio, and File behind a unified interface.

This architecture enables developers to build stream processors that can produce or consume data from any supported backend based on user configuration or environment. For example, a developer can generate data locally using the Stdio backend and later stream it to a production Kafka cluster simply by changing the connection URI.

Sources: sea-streamer-socket/src/lib.rs:1-15, README.md:100-112, src/lib.rs:207-210

Architecture and Components

The Socket API sits between the abstract traits defined in sea-streamer-types and the specific backend implementations. It provides concrete "wrapper" types that dispatch calls to the underlying backend implementation selected during the connection phase.

Core Components

Component	Description
SeaStreamer	The main entry point used to establish connections to a backend.
SeaProducer	A concrete type for sending messages to a stream, wrapping backend-specific producers.
SeaConsumer	A concrete type for receiving messages from a stream, wrapping backend-specific consumers.
SeaMessage	A unified message structure containing metadata (timestamp, sequence, etc.) and payload.
Backend	An internal enumeration representing the active streaming technology (Kafka, Redis, etc.).

Sources: sea-streamer-socket/src/lib.rs:65-73, README.md:114-118

System Flow

The following diagram illustrates how the SeaStreamer selects a backend based on the URI and creates associated producers or consumers.

The URI scheme determines which specialized crate (e.g., sea-streamer-kafka) is instantiated under the SeaStreamer wrapper.

Sources: sea-streamer-socket/src/lib.rs:32-55, README.md:114-125

Backend Dispatch Mechanism

The module uses an internal Backend enum to manage the lifecycle of different implementations. This enum is used by SeaStreamer, SeaProducer, and SeaConsumer to route trait-like methods (such as send or next) to the appropriate sub-crate.

Sequence of Connection and Interaction

The following sequence diagram shows the interaction between the application, the Socket API, and the specific backends during a typical streaming session.

This mechanism ensures that the application code remains identical regardless of whether it is communicating with Kafka, Redis, or a local File.

Sources: sea-streamer-socket/src/lib.rs:1-15, README.md:38-55, sea-streamer-socket/Cargo.toml:21-25

Supported Backends

The Socket API integrates several specialized backends, each enabled via Cargo features.

1. Kafka / Redpanda: Wrapped via sea-streamer-kafka, providing async wrappers for rdkafka.
2. Redis: Wrapped via sea-streamer-redis, abstracting Redis Streams into a producer/consumer model.
3. Stdio: Wrapped via sea-streamer-stdio, allowing streams to be piped through Unix standard I/O.
4. File: Wrapped via sea-streamer-file, supporting real-time streaming and replay from .ss files.

Sources: README.md:127-145, README.md:200-210, sea-streamer-socket/Cargo.toml:26-30

Feature Configuration

Feature	Dependency	Description
backend-kafka	sea-streamer-kafka	Enables Kafka/Redpanda support.
backend-redis	sea-streamer-redis	Enables Redis Streams support.
backend-stdio	sea-streamer-stdio	Enables Standard I/O support (Default).
backend-file	sea-streamer-file	Enables Binary File support.

Sources: sea-streamer-socket/Cargo.toml:26-30, README.md:114-125

Usage Example

The following code demonstrates how the SeaStreamer type is used to connect to any backend and create a consumer.

// The same code works for "kafka://localhost:9092" or "redis://localhost:6379"
let streamer = SeaStreamer::connect(uri, Default::default()).await?;

let mut options = SeaConsumerOptions::new(ConsumerMode::RealTime);
let consumer: SeaConsumer = streamer
    .create_consumer(&[StreamKey::new("my_topic")], options)
    .await?;

while let Ok(message) = consumer.next().await {
    println!("Received: {}", message.message().as_str()?);
}

Sources: README.md:40-60, src/lib.rs:125-145

Conclusion

The Backend-Agnostic Socket API is a critical component of the SeaStreamer toolkit, providing the necessary glue to make stream processing logic portable. By abstracting the complexities of different streaming protocols into concrete, dispatch-oriented types, it enables a "write once, stream anywhere" developer experience. This is particularly valuable for development workflows where local testing uses Stdio or Files while production environments utilize distributed systems like Kafka or Redis.

Sources: sea-streamer-socket/src/lib.rs:1-15, README.md:100-112