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, BlockRuntime, InputPort, Message, PortDescriptor};

/// 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, _runtime: &dyn BlockRuntime) -> 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, BlockRuntime, InputPort, Message, OutputPort, Port, PortDescriptor};
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, runtime: &dyn BlockRuntime) -> Result<(), BlockError> {
        while let Some(message) = self.input.receive()? {
            runtime.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 system = System::new();

let constant = system.block(Const {
    output: OutputPort::new(&system),
    value: 42,
});
let blackhole = system.block(Drop(InputPort::new(&system)));

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