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::*;

Wiring up a system or subsystem

use protoflow::blocks::{Const, Drop};
use protoflow::transports::MockTransport;
use protoflow::System;

let system = System::<MockTransport>::build(|s| {
    let source = s.block(Const::<i32>::new(s.output(), 42));
    let sink = s.block(Drop::<i32>::new(s.input()));
    s.connect(&source.output, &sink.input);
});

Executing a system or subsystem

use protoflow::runtimes::StdRuntime;
use protoflow::transports::MockTransport;
use protoflow::{Runtime, System};

let system = System::<MockTransport>::build(|s| {
    /* ... build the system here ... */
});

let transport = MockTransport::new();
let mut runtime = StdRuntime::new(transport).unwrap();
let running_system = runtime.execute(system).unwrap();

Authoring a trivial function block

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

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

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

Authoring a simple DROP block

use protoflow::derive::Block;
use protoflow::{Block, BlockResult, BlockRuntime, InputPort, Message};

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

impl<T: Message> Block for Drop<T> {
    fn execute(&mut self, _runtime: &dyn BlockRuntime) -> BlockResult {
        while let Some(message) = self.0.recv()? {
            drop(message);
        }
        Ok(())
    }
}

Authoring a simple DELAY block

use protoflow::derive::Block;
use protoflow::{Block, BlockResult, BlockRuntime, InputPort, Message, OutputPort, Port};
use std::time::Duration;

/// A block that passes messages through while delaying them by a fixed
/// duration.
#[derive(Block, Clone)]
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 + Clone + 'static> Block for Delay<T> {
    fn execute(&mut self, runtime: &dyn BlockRuntime) -> BlockResult {
        while let Some(message) = self.input.recv()? {
            runtime.sleep_for(self.delay)?;

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

📚 Reference

Glossary

Blocks

• Buffer
• Const
• Count
• Delay
• Drop
• Random

Features

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

👨‍💻 Development

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