Flow blocks wip

README.md

@@ -114,6 +114,7 @@ The built-in blocks provided by Protoflow are listed below:
 |:------------------|:-------------------------------------------------------------------------------------------------------------------------------|
 | [`Buffer`]        | Stores all messages it receives.                                                                                               |
 | [`ConcatStrings`] | Concatenates the received string messages, with an optional delimiter string inserted between each message.                    |
+| [`Concat`]        | Merges multiple input message streams into a single output stream.                        |
 | [`Const`]         | Sends a constant value.                                                                                                        |
 | [`Count`]         | Counts the number of messages it receives, while optionally passing them through.                                              |
 | [`Decode`]        | Decodes messages from a byte stream.                                                                                           |
@@ -133,6 +134,9 @@ The built-in blocks provided by Protoflow are listed below:
 | [`ReadFile`]      | Reads bytes from the contents of a file.                                                                                       |
 | [`ReadSocket`]    | Reads bytes from a TCP socket.                                                                                                 |
 | [`ReadStdin`]     | Reads bytes from standard input (aka stdin).                                                                                   |
+| [`Replicate`]     | Duplicates a single input message stream into multiple identical output streams.                                               |
+| [`Sort`]          | Sorts a single input message stream in ascending order.                                                                        |
+| [`Split`]         | Divides a single input message stream into multiple output streams using a round-robin approach.                               |
 | [`SplitString`]   | Splits the received input message, with an optional delimiter string parameter.                                                |
 | [`WriteFile`]     | Writes or appends bytes to the contents of a file.                                                                             |
 | [`WriteSocket`]   | Writes bytes to a TCP socket                                                                                                   |
@@ -618,6 +622,32 @@ block-beta
 protoflow execute ReadStdin < input.txt
 ```
 
+#### [`Split`]
+
+Divides a single input message stream into multiple output streams using a round-robin approach.
+
+```mermaid
+block-beta
+    columns 7
+    space:5 Sink1 space:1
+    space:1 Source space:1 Split space:3
+    space:5 Sink2 space:1
+    Source-- "input" -->Split
+    Split-- "output_1" -->Sink1
+    Split-- "output_2" -->Sink2
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Split block
+    class Source hidden
+    class Sink1 hidden
+    class Sink2 hidden
+```
+
+```bash
+protoflow execute Split
+```
+
 #### [`SplitString`]
 
 A block that splits the received input message, with an optional delimiter string parameter
@@ -795,6 +825,7 @@ To add a new block type implementation, make sure to examine and amend:
 [`examples`]: lib/protoflow/examples
 
 [`Buffer`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Buffer.html
+[`Concat`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Concat.html
 [`ConcatStrings`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.ConcatStrings.html
 [`Const`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Const.html
 [`Count`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Count.html
@@ -815,6 +846,9 @@ To add a new block type implementation, make sure to examine and amend:
 [`ReadFile`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.ReadFile.html
 [`ReadSocket`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.ReadSocket.html
 [`ReadStdin`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.ReadStdin.html
+[`Replicate`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Replicate.html
+[`Sort`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Sort.html
+[`Split`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Split.html
 [`SplitString`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.SplitString.html
 [`WriteFile`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.WriteFile.html
 [`WriteSocket`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.WriteSocket.html

lib/protoflow-blocks/doc/flow/split.mmd

@@ -0,0 +1,15 @@
+block-beta
+    columns 7
+    space:5 Sink1 space:1
+    space:1 Source space:1 Split space:3
+    space:5 Sink2 space:1
+    Source-- "input" -->Split
+    Split-- "output_1" -->Sink1
+    Split-- "output_2" -->Sink2
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Split block
+    class Source hidden
+    class Sink1 hidden
+    class Sink2 hidden

lib/protoflow-blocks/doc/flow/split.seq.mmd

@@ -0,0 +1,29 @@
+sequenceDiagram
+    autonumber
+    participant BlockA as Another block
+    participant Split.input as Split.input port
+    participant Split as Split block
+    participant Split.output_1 as Split.output_1 port
+    participant BlockB as Another block
+    participant Split.output_2 as Split.output_2 port
+    participant BlockC as Another block
+
+    BlockA-->>Split: Connect
+    Split-->>BlockB: Connect
+    Split-->>BlockC: Connect
+
+    loop Split process
+        BlockA->>Split: Message
+        Split->>Split: Decide the next output port
+        Split->>BlockB: Message
+        BlockA->>Split: Message
+        Split->>Split: Decide the next output port
+        Split->>BlockC: Message
+    end
+
+    BlockA-->>Split: Disconnect
+    Split-->>Split.input: Close
+    Split-->>Split.output_1: Close
+    Split-->>BlockB: Disconnect
+    Split-->>Split.output_2: Close
+    Split-->>BlockC: Disconnect

lib/protoflow-blocks/src/block_config.rs

@@ -30,12 +30,6 @@ pub enum BlockConfig {
         feature = "hash-sha1",
         feature = "hash-sha2"
     ))]
-    #[cfg(any(
-        feature = "hash-blake3",
-        feature = "hash-md5",
-        feature = "hash-sha1",
-        feature = "hash-sha2"
-    ))]
     Hash(HashBlockConfig),
     Io(IoBlockConfig),
     Math(MathBlockConfig),
@@ -63,6 +57,12 @@ impl<'de> serde::Deserialize<'de> for BlockConfig {
                         .unwrap()
                 }
 
+                "Concat" | "Replicate" | "Split" | "Sort" => {
+                    FlowBlockConfig::deserialize(value.clone())
+                        .map(BlockConfig::Flow)
+                        .unwrap()
+                }
+
                 #[cfg(any(
                     feature = "hash-blake3",
                     feature = "hash-md5",

lib/protoflow-blocks/src/block_tag.rs

@@ -18,6 +18,10 @@ pub enum BlockTag {
     Drop,
     Random,
     // FlowBlocks
+    Concat,
+    Replicate,
+    Sort,
+    Split,
     // HashBlocks
     #[cfg(any(
         feature = "hash-blake3",
@@ -73,6 +77,7 @@ impl BlockTag {
         use BlockTag::*;
         match self {
             Buffer => "Buffer",
+            Concat => "Concat",
             Const => "Const",
             Count => "Count",
             Delay => "Delay",
@@ -101,6 +106,9 @@ impl BlockTag {
             ReadSocket => "ReadSocket",
             #[cfg(feature = "std")]
             ReadStdin => "ReadStdin",
+            Replicate => "Replicate",
+            Sort => "Sort",
+            Split => "Split",
             #[cfg(feature = "std")]
             WriteFile => "WriteFile",
             #[cfg(feature = "std")]
@@ -124,6 +132,7 @@ impl FromStr for BlockTag {
         use BlockTag::*;
         Ok(match input {
             "Buffer" => Buffer,
+            "Concat" => Concat,
             "Const" => Const,
             "Count" => Count,
             "Delay" => Delay,
@@ -152,6 +161,9 @@ impl FromStr for BlockTag {
             "ReadSocket" => ReadSocket,
             #[cfg(feature = "std")]
             "ReadStdin" => ReadStdin,
+            "Replicate" => Replicate,
+            "Sort" => Sort,
+            "Split" => Split,
             #[cfg(feature = "std")]
             "WriteFile" => WriteFile,
             #[cfg(feature = "std")]
@@ -186,6 +198,7 @@ impl BlockInstantiation for BlockTag {
         use BlockTag::*;
         match self {
             Buffer => Box::new(super::Buffer::<Any>::with_system(system)),
+            Concat => Box::new(super::Concat::<Any>::with_system(system)),
             Const => Box::new(super::Const::<String>::with_system(system, String::new())),
             Count => Box::new(super::Count::<Any>::with_system(system)),
             Delay => Box::new(super::Delay::<Any>::with_system(system, None)),
@@ -214,6 +227,9 @@ impl BlockInstantiation for BlockTag {
             ReadSocket => Box::new(super::ReadSocket::with_system(system, None)),
             #[cfg(feature = "std")]
             ReadStdin => Box::new(super::ReadStdin::with_system(system, None)),
+            Replicate => Box::new(super::Replicate::<Any>::with_system(system)),
+            Sort => Box::new(super::Sort::<Any>::with_system(system)),
+            Split => Box::new(super::Split::<Any>::with_system(system)),
             #[cfg(feature = "std")]
             WriteFile => Box::new(super::WriteFile::with_system(system, None)),
             #[cfg(feature = "std")]

lib/protoflow-blocks/src/blocks/flow.rs

@@ -1,30 +1,138 @@
 // This is free and unencumbered software released into the public domain.
 
 pub mod flow {
+    use crate::{InputPortName, OutputPortName};
+
     use super::{
-        prelude::{Cow, Named},
-        BlockConnections, BlockInstantiation,
+        prelude::{vec, Box, Cow, Named, Vec},
+        BlockConnections, BlockInstantiation, System,
     };
 
-    pub trait FlowBlocks {}
+    use protoflow_core::{Block, Message};
+
+    pub trait FlowBlocks {
+        fn concat<T: Message + Into<T> + 'static>(&mut self) -> Concat<T>;
+        fn replicate<T: Message + Into<T> + 'static>(&mut self) -> Replicate<T>;
+        fn sort<T: Message + Into<T> + PartialOrd + 'static>(&mut self) -> Sort<T>;
+        fn split<T: Message + Into<T> + 'static>(&mut self) -> Split<T>;
+    }
 
     #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
     #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
-    pub enum FlowBlockTag {}
+    pub enum FlowBlockTag {
+        Concat,
+        Replicate,
+        Sort,
+        Split,
+    }
 
     #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
     #[derive(Clone, Debug)]
-    pub enum FlowBlockConfig {}
+    pub enum FlowBlockConfig {
+        Concat {
+            input_1: InputPortName,
+            input_2: InputPortName,
+            output: OutputPortName,
+        },
+        Replicate {
+            input: InputPortName,
+            output_1: OutputPortName,
+            output_2: OutputPortName,
+        },
+        Sort {
+            input: InputPortName,
+            stop: InputPortName,
+            output: OutputPortName,
+        },
+        Split {
+            input: InputPortName,
+            output_1: OutputPortName,
+            output_2: OutputPortName,
+        },
+    }
 
     impl Named for FlowBlockConfig {
         fn name(&self) -> Cow<str> {
-            unreachable!()
+            use FlowBlockConfig::*;
+            Cow::Borrowed(match self {
+                Concat { .. } => "Concat",
+                Replicate { .. } => "Replicate",
+                Sort { .. } => "Sort",
+                Split { .. } => "Split",
+            })
         }
     }
 
-    impl BlockConnections for FlowBlockConfig {}
+    impl BlockConnections for FlowBlockConfig {
+        fn output_connections(&self) -> Vec<(&'static str, Option<OutputPortName>)> {
+            use FlowBlockConfig::*;
+            match self {
+                Concat { output, .. } => {
+                    vec![("output", Some(output.clone()))]
+                }
+                Replicate {
+                    output_1, output_2, ..
+                } => {
+                    vec![
+                        ("output_1", Some(output_1.clone())),
+                        ("output_2", Some(output_2.clone())),
+                    ]
+                }
+                Sort { output, .. } => {
+                    vec![("output", Some(output.clone()))]
+                }
+                Split {
+                    output_1, output_2, ..
+                } => {
+                    vec![
+                        ("output_1", Some(output_1.clone())),
+                        ("output_2", Some(output_2.clone())),
+                    ]
+                }
+            }
+        }
+    }
+
+    impl BlockInstantiation for FlowBlockConfig {
+        fn instantiate(&self, system: &mut System) -> Box<dyn Block> {
+            use super::SystemBuilding;
+            use FlowBlockConfig::*;
+            match self {
+                Concat { .. } => Box::new(super::Concat::new(
+                    system.input_any(),
+                    system.input_any(),
+                    system.output(),
+                )),
+                Replicate { .. } => Box::new(super::Replicate::new(
+                    system.input_any(),
+                    system.output(),
+                    system.output(),
+                )),
+                Sort { .. } => Box::new(super::Sort::new(
+                    system.input_any(),
+                    system.input(),
+                    system.output(),
+                )),
+                Split { .. } => Box::new(super::Split::new(
+                    system.input_any(),
+                    system.output(),
+                    system.output(),
+                )),
+            }
+        }
+    }
+
+    mod concat;
+    pub use concat::*;
+
+    mod replicate;
+    pub use replicate::*;
+
+    mod sort;
+    pub use sort::*;
 
-    impl BlockInstantiation for FlowBlockConfig {}
+    mod split;
+    pub use split::*;
 }
 
 pub use flow::*;