StreamWeave

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:

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:

🚀 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:

• database - Generic database support
• database-mysql - MySQL integration
• database-postgresql - PostgreSQL integration
• database-sqlite - SQLite 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:

• integrations/opentelemetry - OpenTelemetry integration
• integrations/sql - SQL query support
• metrics - Metrics collection and Prometheus integration
• visualization - Pipeline and graph visualization

📚 Documentation

• API Documentation - Full API reference on docs.rs
• Local Documentation - Generated with rustdoc (run ./bin/docs)
• Graph API Guide - Advanced graph patterns, routing strategies, and Flow-Based Programming
• Getting Started Guide
• Architecture Overview
• Common Use Cases
• Troubleshooting
• Contributing Guide

📖 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