Protoflow

🚧 This is presently under heavy construction.

🛠️ Prerequisites

• Rust 1.70+

⬇️ Installation

Installation via Cargo

$ cargo add protoflow

👉 Examples

Importing the library

use protoflow::*;
use protoflow::derive::*;

Authoring a simple DROP block

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

/// A block that simply discards all messages it receives.
#[derive(Block)]
pub struct Drop<T: Message>(#[input] pub InputPort<T>);

impl<T: Message> Block for Drop<T> {
    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::derive::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.
#[derive(Block)]
pub struct Delay<T: Message> {
    /// The input message stream.
    #[input]
    pub input: InputPort<T>,
    /// The output target for the stream being passed through.
    #[output]
    pub output: OutputPort<T>,
    /// A configuration parameter for how much delay to add.
    #[parameter]
    pub delay: Duration,
}

impl<T: Message> Block for Delay<T> {
    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(())
    }
}

Authoring a trivial function block

use protoflow::derive::FunctionBlock;
use protoflow::{BlockError, FunctionBlock, InputPort, OutputPort};

/// A block that simply echoes inputs to outputs.
#[derive(FunctionBlock)]
pub struct Echo(pub InputPort<i64>, pub OutputPort<i64>);

impl FunctionBlock<i64, i64> for Echo {
    fn compute(&self, input: i64) -> Result<i64, BlockError> {
        Ok(input)
    }
}

Wiring up a system or subsystem

use protoflow::blocks::{Const, Drop};
use protoflow::{InputPort, OutputPort, System};

let mut system = System::new();

let constant = system.block(Const {
    output: OutputPort::default(),
    value: 42,
});

let blackhole = system.block(Drop(InputPort::default()));

system.connect(&constant.output, &blackhole.0)?;

📚 Reference

Blocks

• Const
• Count
• Delay
• Drop
• Random

Transports

• Flume
• ZeroMQ

Runtimes

• StdThread
• Tokio
• Web

Features

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

👨‍💻 Development

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