Examples: Price Feed & Sea-ORM Sink

Relevant source files

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

• examples/sea-orm-sink/src/main.rs
• examples/sea-orm-sink/README.md
• examples/Cargo.toml
• README.md
• src/lib.rs
• sea-streamer-socket/src/lib.rs

The Price Feed and Sea-ORM Sink examples demonstrate the practical application of the SeaStreamer toolkit in building real-time data pipelines. These examples showcase how to stream data from a producer (such as a price feed) and consume it using a sink that persists the information into a relational database via SeaORM.

The workflow typically involves running a price feed producer that publishes JSON messages to a stream (e.g., Redis or Kafka), followed by the Sea-ORM Sink which subscribes to these messages, deserializes them, and performs database insertions.

Sources: examples/sea-orm-sink/README.md:3-8, README.md:65-95

Sea-ORM Sink Architecture

The Sea-ORM Sink is a stream consumer designed to bridge real-time streams with persistent SQL storage. It utilizes sea-streamer for backend-agnostic stream consumption and sea-orm for database abstraction.

Component Overview

Component	Responsibility
SeaStreamer	Manages connection to the streaming backend (Redis/Kafka) and creates consumers.
SeaConsumer	Subscribes to specific stream keys and retrieves incoming messages asynchronously.
SeaORM	Handles database connections, schema generation, and ORM mapping for data persistence.
JSON Deserializer	Converts raw stream payloads into structured Rust data models using serde_json.

Sources: examples/sea-orm-sink/src/main.rs:24-52, examples/sea-orm-sink/README.md:3-5

Data Flow Logic

The data flow follows a linear path from the stream to the database:

1. Initialization: The application parses CLI arguments for the Streamer URI and Stream Key.
2. Database Setup: It connects to a SQLite database (defaulting to the stream key name) and automatically creates the necessary tables if they do not exist.
3. Stream Connection: A connection is established to the SeaStreamer backend.
4. Processing Loop: The application enters an infinite loop, pulling the next message from the consumer, deserializing the JSON payload, and saving it to the database.

The diagram shows the sequential execution and the infinite processing loop of the Sea-ORM sink.

Sources: examples/sea-orm-sink/src/main.rs:32-52, examples/sea-orm-sink/README.md:5-10

Implementation Details

Database Schema Generation

The sink uses sea_orm::Schema to dynamically build and execute table creation statements based on the defined entities.

async fn create_tables(db: &DbConn) -> Result<(), DbErr> {
    let builder = db.get_database_backend();
    let schema = Schema::new(builder);

    let stmt = builder.build(
        schema
            .create_table_from_entity(spread::Entity)
            .if_not_exists(),
    );
    db.execute(stmt).await?;
    Ok(())
}

Sources: examples/sea-orm-sink/src/main.rs:55-68

Message Processing

Messages are received as SeaMessage objects. The payload is extracted as a string and mapped to a local Item struct, which contains the model to be persisted.

loop {
    let message = consumer.next().await?;
    let payload = message.message();
    let json = payload.as_str()?;
    let item: Item = serde_json::from_str(json)?;
    let mut spread = item.spread.into_active_model();
    spread.id = NotSet;
    spread.save(&db).await?;
}

Sources: examples/sea-orm-sink/src/main.rs:44-52

Integration with Price Feed

The Sea-ORM Sink is designed to work in tandem with the price-feed example. The price feed acts as the producer, generating financial data (e.g., GBP_USD spreads) and publishing them to the configured streamer.

Execution Example

To run the full pipeline, users typically initialize a Redis instance, start the price feed, and then start the sink:

# Start the sink (subscribes to GBP_USD by default)
RUST_LOG=info cargo run --package sea-streamer-sea-orm-sink

The sink creates a local SQLite file (e.g., GBP_USD.sqlite) to store the incoming data.

Sources: examples/sea-orm-sink/README.md:3-12, README.md:120-130

Summary

The Price Feed and Sea-ORM Sink examples provide a blueprint for building micro-service oriented stream processors. By combining sea-streamer's backend-agnostic socket API with sea-orm's database capabilities, developers can create self-contained, low-resource processors that move data seamlessly from real-time streams to relational storage without recompilation for different backends.

Sources: README.md:38-45, sea-streamer-socket/src/lib.rs:10-18