streamweave

Core traits and types for StreamWeave
The foundational abstractions that power all StreamWeave data processing.

The streamweave package provides the core traits and types that form the foundation of the StreamWeave framework. All other StreamWeave packages depend on these core abstractions to build producers, transformers, and consumers.

✨ Key Features

Producer Trait: Define components that generate data streams
Transformer Trait: Define components that transform data streams
Consumer Trait: Define components that consume data streams
Input/Output Traits: Type-safe stream interfaces
Port System: Type-safe multi-port connections for graph-based processing
Configuration System: Unified configuration for error handling and component naming
Error Handling Integration: Seamless integration with streamweave-error

📦 Installation

Add this to your Cargo.toml:

[dependencies]
streamweave = "0.3.0"

🚀 Quick Start

Basic Pipeline Example

use streamweave::{
    Producer, Transformer, Consumer,
    Input, Output,
};

// This example shows the core traits in action
// See specific package implementations for concrete examples

For a complete working example, see the pipeline package or check out the examples directory.

📖 API Overview

Producer Trait

The Producer trait defines components that generate data streams. Producers are the starting point of any StreamWeave pipeline.

use streamweave::Producer;

#[async_trait::async_trait]
trait Producer: Output {
    type OutputPorts: PortList;
    
    fn produce(&mut self) -> Self::OutputStream;
    
    // Configuration methods
    fn with_config(&self, config: ProducerConfig<Self::Output>) -> Self;
    fn with_name(self, name: String) -> Self;
    
    // Error handling
    fn handle_error(&self, error: &StreamError<Self::Output>) -> ErrorAction;
}

Key Methods:

produce() - Generates the output stream
with_config() - Applies configuration (error strategy, name)
handle_error() - Handles errors according to configured strategy

Example Producer Implementation:

use streamweave::{Producer, Output, ProducerConfig};
use streamweave_error::ErrorStrategy;
use futures::Stream;
use std::pin::Pin;

struct NumberProducer {
    numbers: Vec<i32>,
    config: ProducerConfig<i32>,
}

impl Output for NumberProducer {
    type Output = i32;
    type OutputStream = Pin<Box<dyn Stream<Item = i32> + Send>>;
}

#[async_trait::async_trait]
impl Producer for NumberProducer {
    type OutputPorts = (i32,);
    
    fn produce(&mut self) -> Self::OutputStream {
        Box::pin(futures::stream::iter(self.numbers.clone()))
    }
    
    fn set_config_impl(&mut self, config: ProducerConfig<Self::Output>) {
        self.config = config;
    }
    
    fn get_config_impl(&self) -> &ProducerConfig<Self::Output> {
        &self.config
    }
    
    fn get_config_mut_impl(&mut self) -> &mut ProducerConfig<Self::Output> {
        &mut self.config
    }
}

Transformer Trait

The Transformer trait defines components that transform data streams. Transformers process items as they flow through the pipeline.

use streamweave::Transformer;

#[async_trait::async_trait]
trait Transformer: Input + Output {
    type InputPorts: PortList;
    type OutputPorts: PortList;
    
    fn transform(&mut self, input: Self::InputStream) -> Self::OutputStream;
    
    // Configuration methods
    fn with_config(&self, config: TransformerConfig<Self::Input>) -> Self;
    fn with_name(self, name: String) -> Self;
    
    // Error handling
    fn handle_error(&self, error: &StreamError<Self::Input>) -> ErrorAction;
}

Key Methods:

transform() - Transforms the input stream into an output stream
with_config() - Applies configuration (error strategy, name)
handle_error() - Handles errors according to configured strategy

Example Transformer Implementation:

use streamweave::{Transformer, Input, Output, TransformerConfig};
use streamweave_error::ErrorStrategy;
use futures::StreamExt;
use std::pin::Pin;

struct DoubleTransformer {
    config: TransformerConfig<i32>,
}

impl Input for DoubleTransformer {
    type Input = i32;
    type InputStream = Pin<Box<dyn Stream<Item = i32> + Send>>;
}

impl Output for DoubleTransformer {
    type Output = i32;
    type OutputStream = Pin<Box<dyn Stream<Item = i32> + Send>>;
}

#[async_trait::async_trait]
impl Transformer for DoubleTransformer {
    type InputPorts = (i32,);
    type OutputPorts = (i32,);
    
    fn transform(&mut self, input: Self::InputStream) -> Self::OutputStream {
        Box::pin(input.map(|x| x * 2))
    }
    
    fn set_config_impl(&mut self, config: TransformerConfig<Self::Input>) {
        self.config = config;
    }
    
    fn get_config_impl(&self) -> &TransformerConfig<Self::Input> {
        &self.config
    }
    
    fn get_config_mut_impl(&mut self) -> &mut TransformerConfig<Self::Input> {
        &mut self.config
    }
}

Consumer Trait

The Consumer trait defines components that consume data streams. Consumers are the end point of a pipeline.

use streamweave::Consumer;

#[async_trait::async_trait]
trait Consumer: Input {
    type InputPorts: PortList;
    
    async fn consume(&mut self, stream: Self::InputStream);
    
    // Configuration methods
    fn with_config(&self, config: ConsumerConfig<Self::Input>) -> Self;
    fn with_name(self, name: String) -> Self;
    
    // Error handling
    fn handle_error(&self, error: &StreamError<Self::Input>) -> ErrorAction;
}

Key Methods:

consume() - Consumes the input stream (async)
with_config() - Applies configuration (error strategy, name)
handle_error() - Handles errors according to configured strategy

Example Consumer Implementation:

use streamweave::{Consumer, Input, ConsumerConfig};
use streamweave_error::ErrorStrategy;
use futures::StreamExt;
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::Mutex;

struct VecConsumer<T> {
    items: Arc<Mutex<Vec<T>>>,
    config: ConsumerConfig<T>,
}

impl<T: Send + Sync + 'static> Input for VecConsumer<T> {
    type Input = T;
    type InputStream = Pin<Box<dyn Stream<Item = T> + Send>>;
}

#[async_trait::async_trait]
impl<T: Send + Sync + 'static> Consumer for VecConsumer<T> {
    type InputPorts = (T,);
    
    async fn consume(&mut self, mut stream: Self::InputStream) {
        while let Some(item) = stream.next().await {
            self.items.lock().await.push(item);
        }
    }
    
    fn set_config_impl(&mut self, config: ConsumerConfig<Self::Input>) {
        self.config = config;
    }
    
    fn get_config_impl(&self) -> &ConsumerConfig<Self::Input> {
        &self.config
    }
    
    fn get_config_mut_impl(&mut self) -> &mut ConsumerConfig<Self::Input> {
        &mut self.config
    }
}

Input and Output Traits

The Input and Output traits define the stream interfaces for components.

Input Trait:

pub trait Input {
    type Input;
    type InputStream: Stream<Item = Self::Input> + Send;
}

Output Trait:

pub trait Output {
    type Output;
    type OutputStream: Stream<Item = Self::Output> + Send;
}

These traits ensure type safety and enable components to be composed together in pipelines and graphs.

Port System

The port system enables type-safe multi-port connections in the Graph API. Ports are represented as tuples, allowing components to have multiple inputs or outputs.

use streamweave::port::{PortList, GetPort};

// Single port
type SinglePort = (i32,);

// Multiple ports
type MultiPort = (i32, String, bool);

// Extract port types at compile time
type FirstPort = <MultiPort as GetPort<0>>::Type;  // i32
type SecondPort = <MultiPort as GetPort<1>>::Type; // String
type ThirdPort = <MultiPort as GetPort<2>>::Type;  // bool

The port system supports up to 12 ports per component, with compile-time type checking.

Configuration System

All components support configuration through ProducerConfig, TransformerConfig, and ConsumerConfig:

use streamweave_error::ErrorStrategy;

// Configure error handling
let config = ProducerConfig::default()
    .with_error_strategy(ErrorStrategy::Skip)
    .with_name("my_producer".to_string());

let producer = producer.with_config(config);

Configuration Options:

error_strategy - How to handle errors (Stop, Skip, Retry, Custom)
name - Component name for logging and metrics

📚 Usage Examples

Creating a Producer

use streamweave::{Producer, Output, ProducerConfig};
use streamweave_error::ErrorStrategy;

// Create a producer with error handling
let producer = MyProducer::new()
    .with_config(
        ProducerConfig::default()
            .with_error_strategy(ErrorStrategy::Retry(3))
            .with_name("data_source".to_string())
    );

Creating a Transformer

use streamweave::{Transformer, TransformerConfig};
use streamweave_error::ErrorStrategy;

// Create a transformer with error handling
let transformer = MyTransformer::new()
    .with_config(
        TransformerConfig::default()
            .with_error_strategy(ErrorStrategy::Skip)
            .with_name("data_processor".to_string())
    );

Creating a Consumer

use streamweave::{Consumer, ConsumerConfig};
use streamweave_error::ErrorStrategy;

// Create a consumer with error handling
let consumer = MyConsumer::new()
    .with_config(
        ConsumerConfig {
            error_strategy: ErrorStrategy::Stop,
            name: "data_sink".to_string(),
        }
    );

Error Handling Strategies

All components support multiple error handling strategies:

use streamweave_error::ErrorStrategy;

// Stop on first error (default)
ErrorStrategy::Stop

// Skip errors and continue processing
ErrorStrategy::Skip

// Retry up to N times
ErrorStrategy::Retry(3)

// Custom error handler
ErrorStrategy::new_custom(|error| {
    // Custom logic
    ErrorAction::Skip
})

🏗️ Architecture

The streamweave core package provides the foundational abstractions:

┌─────────────┐
│   Producer  │───produces───> Stream<T>
└─────────────┘
       │
       │ Stream flows through
       ▼
┌─────────────┐
│ Transformer │───transforms───> Stream<U>
└─────────────┘
       │
       │ Stream flows through
       ▼
┌─────────────┐
│  Consumer   │───consumes───> (writes, stores, etc.)
└─────────────┘

All components:

Implement Input and/or Output traits for type safety
Support configuration for error handling and naming
Integrate with the error handling system
Can be used in both Pipeline and Graph APIs

🔗 Dependencies

streamweave depends on:

tokio - Async runtime
futures - Stream abstractions
async-trait - Async trait support
chrono - Timestamp support
streamweave-error - Error handling system

🎯 Use Cases

The core traits are used to:

Build Custom Components: Create producers, transformers, and consumers for specific use cases
Type-Safe Composition: Ensure components can be safely connected in pipelines
Error Handling: Provide consistent error handling across all components
Graph API: Enable multi-port connections in complex topologies
Configuration: Standardize component configuration and naming

🔍 Error Handling

All components integrate with the streamweave-error package for consistent error handling:

Error Strategies: Stop, Skip, Retry, or Custom handlers
Error Context: Automatic error context creation with timestamps and component info
Component Info: Automatic component identification for error reporting

⚡ Performance Considerations

Zero-Cost Abstractions: Traits compile to efficient code with no runtime overhead
Stream-Based: All processing is stream-based for memory efficiency
Async: Full async/await support for concurrent processing
Type Safety: Compile-time type checking prevents runtime errors

📝 Examples

For more examples, see:

Pipeline Examples
Graph Examples
Package Implementations - See specific packages for concrete implementations

📖 Documentation

🔗 See Also

streamweave-error - Error handling system
streamweave-pipeline - Pipeline builder and execution
streamweave-graph - Graph API for complex topologies
streamweave-message - Message envelope and metadata

🤝 Contributing

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

📄 License

This project is licensed under the CC BY-SA 4.0 license.

streamweave 0.4.0

streamweave

✨ Key Features

📦 Installation

🚀 Quick Start

Basic Pipeline Example

📖 API Overview

Producer Trait

Transformer Trait

Consumer Trait

Input and Output Traits

Port System

Configuration System

📚 Usage Examples

Creating a Producer

Creating a Transformer

Creating a Consumer

Error Handling Strategies

🏗️ Architecture

🔗 Dependencies

🎯 Use Cases

🔍 Error Handling

⚡ Performance Considerations

📝 Examples

📖 Documentation

🔗 See Also

🤝 Contributing

📄 License