streamweave 0.8.2

Composable, async, stream-first computation in pure Rust
Documentation

StreamWeave

Crates.io Documentation CI codecov License: CC BY-SA 4.0

Composable, async, stream-first computation in pure Rust
Build fully composable, async data pipelines using a fluent API.

StreamWeave is a general-purpose Rust framework built around the concept of streaming data, with a focus on simplicity, composability, and performance.

High-Performance Streaming: Process 2-6 million messages per second with in-process zero-copy execution. Perfect for high-throughput data processing pipelines.

✨ Key Features

  • Pure Rust API with zero-cost abstractions
  • Full async/await compatibility via futures::Stream
  • Fluent pipeline-style API with type-safe builder pattern
  • Graph-based API for complex topologies with fan-in/fan-out patterns
  • Flow-Based Programming (FBP) patterns with type-safe routing
  • Comprehensive error handling system with multiple strategies
  • Code-as-configuration — no external DSLs
  • Extensive package ecosystem for I/O, transformations, and integrations

📦 Core Concepts

StreamWeave breaks computation into three primary building blocks:

Component Description
Producer Starts a stream of data
Transformer Transforms stream items (e.g., map/filter)
Consumer Consumes the stream, e.g. writing, logging

All components can be chained together fluently. These components can be used in both the Pipeline API (for simple linear flows) and the Graph API (for complex topologies with fan-in/fan-out patterns).

🔀 Pipeline vs Graph API

StreamWeave provides two APIs for building data processing workflows:

Feature Pipeline API Graph API
Use Case Simple linear flows Complex topologies
Topology Single path: Producer → Transformer → Consumer Multiple paths, fan-in/fan-out
Routing Sequential processing Configurable routing strategies
Complexity Lower complexity, easier to use Higher flexibility, more powerful
Best For ETL pipelines, simple transformations Complex workflows, parallel processing, data distribution

🚀 Quick Start

Installation

Add StreamWeave to your Cargo.toml:

[dependencies]
streamweave = "0.8.0"

Basic Example

use streamweave::PipelineBuilder;
use streamweave_array::ArrayProducer;
use streamweave_transformers::MapTransformer;
use streamweave_vec::VecConsumer;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pipeline = PipelineBuilder::new()
        .producer(ArrayProducer::new(vec![1, 2, 3, 4, 5]))
        .transformer(MapTransformer::new(|x: i32| x * 2))
        .consumer(VecConsumer::new());

    let ((), result) = pipeline.run().await?;
    println!("Result: {:?}", result.collected);
    Ok(())
}

For more examples and detailed documentation, see the package documentation below.

📦 Packages

StreamWeave is organized as a monorepo with 39 packages, each providing specific functionality. Each package has its own README with detailed documentation, examples, and API reference.

Core Foundation Packages

These are the foundational packages that other packages depend on:

  • streamweave - Core traits and types (Producer, Transformer, Consumer)
  • error - Error handling system with multiple strategies
  • message - Message envelope and metadata
  • offset - Offset management for exactly-once processing
  • transaction - Transaction support and boundaries

System Packages

Core system functionality:

  • pipeline - Pipeline builder and execution
  • graph - Graph API for complex topologies
  • stateful - Stateful processing and state management
  • window - Windowing operations (tumbling, sliding, session)

I/O Packages

Standard I/O and file system operations:

  • stdio - Standard input/output streaming
  • file - File I/O operations
  • fs - File system operations and directory traversal
  • tempfile - Temporary file handling
  • path - Path manipulation and transformations

Data Format Packages

Data format parsing and serialization:

  • csv - CSV parsing and writing
  • jsonl - JSON Lines format support
  • parquet - Parquet format support

Database Packages

Database integration:

Network Packages

Network protocol integration:

  • kafka - Apache Kafka producer and consumer
  • redis - Redis Streams integration
  • http-server - HTTP graph server with Axum integration

Producer/Consumer Packages

Various data source and sink implementations:

  • array - Array-based streaming
  • vec - Vector-based streaming
  • env - Environment variable streaming
  • command - Command execution and output streaming
  • process - Process management and monitoring
  • signal - Unix signal handling
  • timer - Time-based and interval-based streaming
  • tokio - Tokio channel integration

Transformers Package

Comprehensive transformer implementations:

  • transformers - All transformer types including:
    • Basic: Map, Filter, Reduce
    • Advanced: Batch, Retry, CircuitBreaker, RateLimit
    • Stateful: RunningSum, MovingAverage
    • Routing: Router, Partition, RoundRobin
    • Merging: Merge, OrderedMerge, Interleave
    • ML: Inference, BatchedInference
    • Utility: Sample, Skip, Take, Limit, Sort, Split, Zip, Timeout, MessageDedupe

Integration and Utility Packages

Observability and integration capabilities:

📚 Documentation

📖 Examples

StreamWeave includes comprehensive examples demonstrating all major features. See the examples directory for:

  • Integration examples (Kafka, Redis, Database, HTTP)
  • File format examples (CSV, JSONL, Parquet)
  • Processing examples (Stateful, Error Handling, Windowing)
  • Visualization examples
  • Graph API examples

Run any example with:

cargo run --example <example_name> --features <required_features>

🤝 Contributing

Contributions are welcome! Please see our Contributing Guide for details.

📄 License

This project is licensed under the Creative Commons Attribution-ShareAlike 4.0 International License.

See [LICENSE](LICENSE) for details.

🙏 Acknowledgments

  • Built with Tokio for async runtime
  • HTTP support powered by Axum
  • Inspired by reactive programming patterns and stream processing frameworks