SeaStreamer

https://www.sea-ql.org/SeaStreamer/
Admin
SeaStreamer is a stream processing toolkit to help you build stream processors in Rust, supporting...
Tokens:8,176
Snippets:128
Trust Score:8.2
Update:3 months ago
Show doc for...
Tokens:
Raw
### Running Kafka Stream Example

Source: https://www.sea-ql.org/SeaStreamer

Demonstrates how to run SeaStreamer with Kafka. This example shows the commands to start a producer, a processor, and a consumer, along with instructions to stop them.

```shell
# Produce some input
cargo run --bin producer -- --stream kafka://localhost:9092/hello1 &
# Start the processor, producing some output
cargo run --bin processor -- --input kafka://localhost:9092/hello1 --output kafka://localhost:9092/hello2 &
# Replay the output
cargo run --bin consumer -- --stream kafka://localhost:9092/hello2
# Remember to stop the processes
kill %1 %2
```

--------------------------------

### Running Redis Stream Example

Source: https://www.sea-ql.org/SeaStreamer

Demonstrates how to run SeaStreamer with Redis. This example shows the commands to start a producer, a processor, and a consumer, along with instructions to stop them.

```shell
# Produce some input
cargo run --bin producer -- --stream redis://localhost:6379/hello1 &
# Start the processor, producing some output
cargo run --bin processor -- --input redis://localhost:6379/hello1 --output redis://localhost:6379/hello2 &
# Replay the output
cargo run --bin consumer -- --stream redis://localhost:6379/hello2
# Remember to stop the processes
kill %1 %2
```

--------------------------------

### Running File and Stdio Stream Example

Source: https://www.sea-ql.org/SeaStreamer

Demonstrates how to run SeaStreamer with file and stdio streams. This example shows creating a file, producing input to it, consuming from it, and processing it to stdio.

```shell
# Create the file
file=/tmp/sea-streamer-$(date +%s)
touch $file && echo "File created at $file"
# Produce some input
cargo run --bin producer -- --stream file://$file/hello &
# Replay the input
cargo run --bin consumer -- --stream file://$file/hello
# Start the processor, producing some output
cargo run --bin processor -- --input file://$file/hello --output stdio:///hello
```

--------------------------------

### Run Basic Processor with File

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

Demonstrates how to run the producer, consumer, and processor examples using local files. The file path is dynamically generated.

```bash
# Create the file  
file=/tmp/sea-streamer-$(date +%s)  
touch $file && echo "File created at $file"  
# Produce some input  
cargo run --bin producer -- --stream file://$file/hello &  
# Replay the input  
cargo run --bin consumer -- --stream file://$file/hello  
# Start the processor, producing some output  
cargo run --bin processor -- --input file://$file/hello --output stdio:///hello  

```

--------------------------------

### Run Basic Processor with Kafka

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

Demonstrates how to run the producer, processor, and consumer examples using Kafka. Remember to stop the processes after use.

```bash
# Produce some input  
cargo run --bin producer -- --stream kafka://localhost:9092/hello1 &  
# Start the processor, producing some output  
cargo run --bin processor -- --input kafka://localhost:9092/hello1 --output kafka://localhost:9092/hello2 &  
# Replay the output  
cargo run --bin consumer -- --stream kafka://localhost:9092/hello2  
# Remember to stop the processes  
kill %1 %2  

```

--------------------------------

### Valid Stdio Message Examples

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Provides examples of correctly formatted messages for stdin, demonstrating various combinations of metadata and payload types. These examples adhere to the specified format for successful parsing.

```plaintext
a plain, raw message  
[2022-01-01T00:00:00] { "payload": "anything" }  
[2022-01-01T00:00:00.123 | my_topic] "a string payload"  
[2022-01-01T00:00:00 | my-topic-2 | 123] ["array", "of", "values"]  
[2022-01-01T00:00:00 | my-topic-2 | 123 | 4] { "payload": "anything" }  
[my_topic] a string payload  
[my_topic | 123] { "payload": "anything" }  
[my_topic | 123 | 4] { "payload": "anything" }  
```

--------------------------------

### Run Basic Processor with Redis

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

Demonstrates how to run the producer, processor, and consumer examples using Redis. Remember to stop the processes after use.

```bash
# Produce some input  
cargo run --bin producer -- --stream redis://localhost:6379/hello1 &  
# Start the processor, producing some output  
cargo run --bin processor -- --input redis://localhost:6379/hello1 --output redis://localhost:6379/hello2 &  
# Replay the output  
cargo run --bin consumer -- --stream redis://localhost:6379/hello2  
# Remember to stop the processes  
kill %1 %2  

```

--------------------------------

### Run Blocking Processor Example

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

This example demonstrates how to run a blocking processor with a clock. It uses a fan-out pattern where tasks are randomly assigned to threads, suitable for blocking I/O or CPU-heavy computations.

```bash
alias clock='cargo run --package sea-streamer-stdio --features=executables --bin clock'  
clock -- --stream clock --interval 333ms | \
cargo run --bin blocking -- --input stdio:///clock --output stdio:///output  
```

--------------------------------

### Async Stream Consumer Example

Source: https://www.sea-ql.org/SeaStreamer

A basic example of an asynchronous stream consumer using SeaStreamer with tokio. It connects to a stream, sets consumer options, and continuously fetches and prints messages.

```rust
#[tokio::main]
async fn main() -> Result<()> {
    env_logger::init();


    let Args { stream } = Args::from_args();


    let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;


    let mut options = SeaConsumerOptions::new(ConsumerMode::RealTime);
    options.set_auto_stream_reset(SeaStreamReset::Earliest);


    let consumer: SeaConsumer = streamer
        .create_consumer(stream.stream_keys(), options)
        .await?;


    loop {
        let mess: SeaMessage = consumer.next().await?;
        println!("[{}] {}", mess.timestamp(), mess.message().as_str()?);
    }
}
```

--------------------------------

### Run Resumable Processor Example

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

Demonstrates running a resumable processor that can be killed and restarted, ensuring at-least-once message processing. This pattern is useful for fault tolerance.

```bash
STREAMER_URI="kafka://localhost:9092"
STREAMER_URI="redis://localhost:6379"
# Produce lots of input
cargo run --bin producer -- --stream $STREAMER_URI/hello1
# Run the processor, but kill it before it can process the entire stream
cargo run --bin resumable -- --input $STREAMER_URI/hello1 --output stdio:///hello2 | head -n 10
cargo run --bin resumable -- --input $STREAMER_URI/hello1 --output stdio:///hello2 | head -n 10
cargo run --bin resumable -- --input $STREAMER_URI/hello1 --output stdio:///hello2 | head -n 10
```

--------------------------------

### Async Stream Producer Example

Source: https://www.sea-ql.org/SeaStreamer

A basic example of an asynchronous stream producer using SeaStreamer with tokio. It connects to a stream and sends 100 messages with a one-second delay between each.

```rust
#[tokio::main]
async fn main() -> Result<()> {
    env_logger::init();


    let Args { stream } = Args::from_args();


    let streamer = SeaStreamer::connect(stream.streamer(), Default::default()).await?;


    let producer: SeaProducer = streamer
        .create_producer(stream.stream_key()?, Default::default())
        .await?;


    for tick in 0..100 {
        let message = format!(r#"tick {tick}""#);
        eprintln!(message);
        producer.send(message)?;
        tokio::time::sleep(Duration::from_secs(1)).await;
    }


    producer.end().await?; // flush


    Ok(())
}
```

--------------------------------

### Async Stream Processor Example

Source: https://www.sea-ql.org/SeaStreamer

A basic example of an asynchronous stream processor using SeaStreamer. It consumes messages from an input stream, processes them, and sends them to an output stream.

```rust
#[tokio::main]
async fn main() -> Result<()> {
    env_logger::init();


    let Args { input, output } = Args::from_args();


    let streamer = SeaStreamer::connect(input.streamer(), Default::default()).await?;
    let options = SeaConsumerOptions::new(ConsumerMode::RealTime);
    let consumer: SeaConsumer = streamer
        .create_consumer(input.stream_keys(), options)
        .await?;


    let streamer = SeaStreamer::connect(output.streamer(), Default::default()).await?;
    let producer: SeaProducer = streamer
        .create_producer(output.stream_key()?, Default::default())
        .await?;


    loop {
        let message: SeaMessage = consumer.next().await?;
        let message = process(message).await?;
        eprintln!(message);
        producer.send(message)?; // send is non-blocking
    }
}
```

--------------------------------

### Run Buffered Processor Example

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

Illustrates running a buffered processor, which is beneficial when dealing with high-frequency input streams and processors with high impedance. It decouples busy loops using a queue to maximize overall throughput.

```bash
alias clock='cargo run --package sea-streamer-stdio --features=executables --bin clock'
clock -- --stream clock --interval 100ms | \
cargo run --bin buffered -- --input stdio:///clock --output stdio:///output
```

--------------------------------

### Replaying Streams from Kafka/Redis

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Shows how to replay messages from Kafka or Redis using the `relay` program with the `--offset start` option. This allows reprocessing historical data from the beginning of the stream.

```bash
relay -- --input redis://localhost:6379/clock --output stdio:///clock --offset start  
relay -- --input kafka://localhost:9092/clock --output stdio:///clock --offset start  
```

--------------------------------

### Invalid Stdio Message Examples

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Illustrates incorrectly formatted messages for stdin, highlighting common errors such as invalid timestamp formats or non-string payloads. These examples should be avoided to ensure proper message processing.

```plaintext
[Jan 1, 2022] { "payload": "anything" }  
[2022-01-01T00:00:00] 12345  
```

--------------------------------

### ConnectOptions

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/streamer

Options for establishing a connection, including network timeout settings.

```APIDOC
## ConnectOptions

### Description
Options for establishing a connection to the streaming server.

### Fields
#### timeout
- **timeout** (Duration) - Optional - Set the default network timeout for all connections.
```

--------------------------------

### Using the File Backend for Communication

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Leverage the File backend for inter-process communication by specifying file paths as input and output. This allows for persistent or shared data storage between processors.

```bash
file=/tmp/sea-streamer-$(date +%s)  
touch $file  
processor_a --output file://$file  
processor_b --input  file://$file  
```

--------------------------------

### Pipe Producer to Processor with Stdio

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/examples

Demonstrates piping the output of a producer directly to the input of a processor using standard input/output streams.

```bash
# Pipe the producer to the processor  
cargo run --bin producer -- --stream stdio:///hello1 | \  
cargo run --bin processor -- --input stdio:///hello1 --output stdio:///hello2  

```

--------------------------------

### Setting up SeaStreamer CLI Programs

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Define aliases for the `clock` and `relay` CLI programs. `clock` generates timestamped messages, and `relay` forwards messages between different backends. Ensure SeaStreamer is built with the necessary features and packages.

```bash
# The `clock` program generate messages in the form of `{ "tick": N }`  
alias clock='cargo run --package sea-streamer-stdio  --features=executables --bin clock'  
# The `relay` program redirect messages from `input` to `output`  
alias relay='cargo run --package sea-streamer-socket --features=executables,backend-kafka,backend-redis --bin relay'  
```

--------------------------------

### Streaming from Stdio to Redis/Kafka

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Demonstrates piping messages from the `clock` program (via stdio) to the `relay` program, which then forwards them to Redis or Kafka. This showcases a common pattern for ingesting data into message brokers.

```bash
# Stdio -> Redis  
clock -- --stream clock --interval 1s | \  
relay -- --input stdio:///clock --output redis://localhost:6379/clock  
# Stdio -> Kafka  
clock -- --stream clock --interval 1s | \  
relay -- --input stdio:///clock --output kafka://localhost:9092/clock  
```

--------------------------------

### Piping Producer to Processor via Stdio

Source: https://www.sea-ql.org/SeaStreamer

Demonstrates piping the output of a SeaStreamer producer directly to a processor using stdio. This is useful for chaining stream processing stages.

```shell
# Pipe the producer to the processor
cargo run --bin producer -- --stream stdio:///hello1 | cargo run --bin processor -- --input stdio:///hello1 --output stdio:///hello2
```

--------------------------------

### Stdio Loopback for Intra-Process Testing

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/intra-process

Execute tests involving multiple processors by enabling Stdio's loopback option. This feeds messages back to consumers in the same process, allowing for quick, dependency-free testing. Use a unique stream key for each test case to aid in diagnosing failures from stdout logs.

```rust
let stream = StreamKey::new("hello")?;
let mut options = StdioConnectOptions::default();
options.set_loopback(true);
let streamer = StdioStreamer::connect(StreamerUri::zero(), options).await?;
let producer = streamer.create_producer(stream.clone(), Default::default()).await?;
let mut consumer = streamer.create_consumer(&[stream.clone()], Default::default()).await?;

for i in 0..5 {
    let mess = format!("{}", i);
    producer.send(mess)?;
}

let seq = collect(&mut consumer, 5).await;
assert_eq!(seq, [0, 1, 2, 3, 4]);

```

--------------------------------

### connect

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/streamer

Establishes a connection to the streaming server. The Streamer implementation does not need to maintain an open connection.

```APIDOC
## connect

### Description
Establish a connection to the streaming server. The `Streamer` implementation does not have to maintain an open connection to the server.

### Method
(Implicitly an async operation, likely a method on a Streamer instance)

### Parameters
None explicitly defined in this section, but may take `ConnectOptions` implicitly or explicitly based on implementation.
```

--------------------------------

### Create Temporary File for File Backend

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/intra-process

A helper function to create a new, unique temporary file for use with the File backend. This ensures no interference with other processes by using a random file name and `create_new(true)` to prevent overwriting existing files.

```rust
use sea_streamer_file::FileId;
use sea_streamer_types::Timestamp;
use std::fs::OpenOptions;

pub fn temp_file(name: &str) -> Result<FileId, std::io::Error> {
    let path = format!("/tmp/{name}");
    let _file = OpenOptions::new()
        .read(true)
        .write(true) // Make sure we have write permission
        .create_new(true) // Fail if this file already exists
        .open(&path)?;

    Ok(FileId::new(path))
}

```

--------------------------------

### Asynchronous Communication with Files

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Enable asynchronous communication between processors using files. One processor appends data to a file, while another tails the file for real-time processing. The `tail -f` command is crucial for monitoring file changes.

```bash
touch stream # set up an empty file  
tail -f stream | processor_b # program b can be spawned anytime  
processor_a >> stream # append to the file  
```

--------------------------------

### Streaming between Redis and Kafka

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Illustrates bidirectional streaming between Redis and Kafka using the `relay` program. This is useful for data synchronization or migration between these two popular message brokers.

```bash
# Redis -> Kafka  
relay -- --input redis://localhost:6379/clock --output kafka://localhost:9092/clock  
# Kafka -> Redis  
relay -- --input kafka://localhost:9092/clock --output redis://localhost:6379/clock  
```

--------------------------------

### create_generic_producer

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/streamer

Creates a producer capable of streaming to any stream.

```APIDOC
## create_generic_producer

### Description
Create a producer that can stream to any stream.

### Method
(Likely a method on a Streamer instance)

### Parameters
None explicitly defined in this section, but the producer created may require stream information upon creation or use.
```

--------------------------------

### Connecting Processors with Unix Pipes

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Use the standard Unix pipe operator `|` to chain processors sequentially. This is suitable for direct, synchronous data flow between processes.

```bash
processor_a | processor_b  
```

--------------------------------

### Stream and Process Messages

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/consumer

Create an async stream from a consumer, process messages using `map`, and collect them. This is useful for batch processing or transforming messages before consumption.

```rust
let items = consumer  
    .stream()  
    .take(num)  
    .map(process_message)  
    .collect::<Vec<_>>()  
    .await  

```

--------------------------------

### create_consumer

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/streamer

Creates a consumer that subscribes to one or more specified streams.

```APIDOC
## create_consumer

### Description
Create a consumer subscribing to the specified streams.

### Method
(Likely a method on a Streamer instance)

### Parameters
- **streams** (array of strings) - Required - The names of the streams to subscribe to.
```

--------------------------------

### Add sea-streamer to Cargo.toml

Source: https://www.sea-ql.org/SeaStreamer/docs/getting-started/configuration

Add the sea-streamer crate to your project's dependencies in Cargo.toml. You must specify the desired backend and async runtime features.

```toml
sea-streamer = { version = "0.3", features = [ <BACKEND>, <ASYNC_RUNTIME> ] }
```

--------------------------------

### Stdio Message Format with Metadata

Source: https://www.sea-ql.org/SeaStreamer/docs/processors/inter-process

Defines the format for messages sent to stdin, including optional timestamp, stream key, sequence, and shard ID. Any valid UTF-8 string can be a payload. Ensure correct timestamp format and literal square brackets for metadata.

```plaintext
[timestamp | stream_key | sequence | shard_id] payload  
```

--------------------------------

### create_producer

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/streamer

Creates a producer for streaming data to a specified stream.

```APIDOC
## create_producer

### Description
Create a producer that streams to the specified stream.

### Method
(Likely a method on a Streamer instance)

### Parameters
- **stream** (string) - Required - The name of the stream to produce to.
```

--------------------------------

### disconnect

Source: https://www.sea-ql.org/SeaStreamer/docs/streamer/streamer

Disconnects from the streaming server, flushing remaining messages and exiting gracefully. This operation must be awaited.

```APIDOC
## disconnect

### Description
Disconnect from the streaming server. The intention is to flush remaining messages and exit gracefully. You have to `await` this operation until it completes. Once you called this method, all producers and consumers created will become unusable.

### Method
(Implicitly an async operation, likely a method on a Streamer instance)

### Parameters
None.
```
Tokens:
Raw