Protoflow

🚧 This is presently under heavy construction.

🛠️ Prerequisites

• Rust 1.70+

⬇️ Installation

Installation via Cargo

$ cargo add protoflow

👉 Examples

Importing the library

use protoflow::*;

Authoring a simple DROP block

use protoflow::{Block, BlockError, InputPort, Message, PortDescriptor, Scheduler};

/// A block that simply discards all messages it receives.
pub struct Drop<T: Message>(InputPort<T>);

impl<T: Message> Block for Drop<T> {
    fn inputs(&self) -> Vec<PortDescriptor> {
        vec![PortDescriptor::from(&self.0)]
    }

    fn outputs(&self) -> Vec<PortDescriptor> {
        vec![] // no output ports
    }

    fn execute(&mut self, _scheduler: &dyn Scheduler) -> Result<(), BlockError> {
        while let Some(message) = self.0.receive()? {
            drop(message);
        }
        Ok(())
    }
}

Authoring a simple DELAY block

use protoflow::{Block, BlockError, InputPort, Message, OutputPort, Port, PortDescriptor, Scheduler};
use std::time::Duration;

/// A block that passes messages through while delaying them by a fixed
/// duration.
pub struct Delay<T: Message> {
    /// The input message stream.
    input: InputPort<T>,
    /// The output target for the stream being passed through.
    output: OutputPort<T>,
    /// The configuration parameter for how much delay to add.
    delay: Duration,
}

impl<T: Message> Block for Delay<T> {
    fn inputs(&self) -> Vec<PortDescriptor> {
        vec![PortDescriptor::from(&self.input)]
    }

    fn outputs(&self) -> Vec<PortDescriptor> {
        vec![PortDescriptor::from(&self.output)]
    }

    fn execute(&mut self, scheduler: &dyn Scheduler) -> Result<(), BlockError> {
        while let Some(message) = self.input.receive()? {
            scheduler.sleep_for(self.delay)?;

            if self.output.is_connected() {
                self.output.send(&message)?;
            }
        }
        Ok(())
    }
}

📚 Reference

Features

• flume
• rand
• std
• tokio
• tracing
• web
• zeromq

Runtimes

• runtimes::StdThread
• runtimes::Tokio
• runtimes::Web

Transports

• transports::Flume
• transports::ZeroMQ

Blocks

TODO

👨‍💻 Development

$ git clone https://github.com/artob/protoflow.git