streamweave-timer

Timer and interval producer for StreamWeave
Generate events at regular intervals for time-based processing.

The streamweave-timer package provides timer and interval producers for StreamWeave. It enables generating events at regular intervals for time-based processing, scheduling, and periodic tasks.

✨ Key Features

• IntervalProducer: Generate events at regular intervals
• Time-Based Processing: Time-based event generation
• Scheduling: Schedule periodic tasks
• Timestamp Events: Emit timestamps at intervals
• Flexible Intervals: Configurable interval durations

📦 Installation

Add this to your Cargo.toml:

[dependencies]
streamweave-timer = "0.3.0"

🚀 Quick Start

Interval Producer

use streamweave_timer::IntervalProducer;
use streamweave_pipeline::PipelineBuilder;
use std::time::Duration;

let producer = IntervalProducer::new(Duration::from_secs(1));

let pipeline = PipelineBuilder::new()
    .producer(producer)
    .consumer(|timestamp: std::time::SystemTime| {
        println!("Tick at {:?}", timestamp);
    });

pipeline.run().await?;

Periodic Processing

use streamweave_timer::IntervalProducer;
use streamweave_pipeline::PipelineBuilder;
use std::time::Duration;

let producer = IntervalProducer::new(Duration::from_secs(5));

let pipeline = PipelineBuilder::new()
    .producer(producer)
    .transformer(|_timestamp: std::time::SystemTime| {
        // Perform periodic task
        "Periodic task executed".to_string()
    })
    .consumer(|msg: String| {
        println!("{}", msg);
    });

pipeline.run().await?;

📖 API Overview

IntervalProducer

Generates events at regular intervals:

pub struct IntervalProducer {
    pub interval: Duration,
    pub config: ProducerConfig<std::time::SystemTime>,
}

Key Methods:

• new(interval) - Create producer with interval duration
• with_error_strategy(strategy) - Set error handling strategy
• with_name(name) - Set component name
• produce() - Generate stream of timestamps

📚 Usage Examples

Periodic Tasks

Execute periodic tasks:

use streamweave_timer::IntervalProducer;
use streamweave_pipeline::PipelineBuilder;
use std::time::Duration;

let producer = IntervalProducer::new(Duration::from_secs(60));

let pipeline = PipelineBuilder::new()
    .producer(producer)
    .consumer(|_timestamp: std::time::SystemTime| {
        // Execute periodic task every minute
        perform_periodic_task();
    });

pipeline.run().await?;

Time-Based Processing

Process events at intervals:

use streamweave_timer::IntervalProducer;
use streamweave_pipeline::PipelineBuilder;
use std::time::Duration;

let producer = IntervalProducer::new(Duration::from_millis(100));

let pipeline = PipelineBuilder::new()
    .producer(producer)
    .transformer(|timestamp: std::time::SystemTime| {
        format!("Event at {:?}", timestamp)
    })
    .consumer(|msg: String| {
        println!("{}", msg);
    });

pipeline.run().await?;

Error Handling

Configure error handling:

use streamweave_timer::IntervalProducer;
use streamweave_error::ErrorStrategy;
use std::time::Duration;

let producer = IntervalProducer::new(Duration::from_secs(1))
    .with_error_strategy(ErrorStrategy::Skip);

🏗️ Architecture

Timer processing flow:

Timer ──> IntervalProducer ──> Stream<SystemTime> ──> Transformer ──> Stream<T> ──> Consumer

Timer Flow:

1. IntervalProducer generates timestamps at intervals
2. Timestamps flow through transformers
3. Consumer processes time-based events
4. Events generated continuously at intervals

🔧 Configuration

Producer Configuration

• Interval: Duration between events
• Error Strategy: Error handling strategy
• Name: Component name for logging

🔍 Error Handling

Timer errors are handled through the error system:

use streamweave_error::ErrorStrategy;

let producer = IntervalProducer::new(Duration::from_secs(1))
    .with_error_strategy(ErrorStrategy::Skip);

⚡ Performance Considerations

• Interval Accuracy: Intervals are approximate
• Resource Usage: Continuous timers consume resources
• Scheduling: Use appropriate intervals for tasks

📝 Examples

For more examples, see:

• Timer Examples

🔗 Dependencies

streamweave-timer depends on:

• streamweave - Core traits
• streamweave-error - Error handling
• streamweave-message (optional) - Message envelope support
• tokio - Async runtime
• futures - Stream utilities

🎯 Use Cases

Timer integration is used for:

1. Periodic Tasks: Execute tasks at regular intervals
2. Scheduling: Schedule periodic operations
3. Time-Based Processing: Process events based on time
4. Monitoring: Monitor systems at intervals

📖 Documentation

• Full API Documentation
• Repository
• StreamWeave Main Documentation

🔗 See Also

• streamweave - Core traits
• streamweave-error - Error handling

🤝 Contributing

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

📄 License

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