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/concat.mmd

@@ -0,0 +1,15 @@
+block-beta
+    columns 7
+    space:1 Source1 space:5
+    space:3 Concat space:1 Sink space:1
+    space:1 Source2 space:5
+    Source1-- "input" -->Concat
+    Source2-- "input" -->Concat
+    Concat-- "output" -->Sink
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Concat block
+    class Source1 hidden
+    class Source2 hidden
+    class Sink hidden

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

@@ -0,0 +1,28 @@
+sequenceDiagram
+    autonumber
+    participant BlockA as Another block
+    participant BlockA2 as Another block
+    participant Concat.input as Concat.input port
+    participant Concat.input2 as Concat.input port
+    participant Concat as Concat block
+    participant Concat.output as Concat.output port
+    participant BlockB as Another block
+
+    BlockA-->>Concat: Connect
+    BlockA2-->>Concat: Connect
+    Concat-->>BlockB: Connect
+
+    loop Concat process
+        BlockA->>Concat: Message
+        Concat->>Concat: Store message
+        BlockA2->>Concat: Message
+        Concat->>Concat: Store message
+    end
+    Concat->>Concat: Concat messages
+    Concat->>BlockB: Message
+    BlockA-->>Concat: Disconnect
+    BlockA2-->>Concat: Disconnect
+    Concat-->>Concat.input: Close
+    Concat-->>Concat.input2: Close
+    Concat-->>Concat.output: Close
+    Concat-->>BlockB: Disconnect

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

@@ -0,0 +1,17 @@
+block-beta
+    columns 7
+    space:5 Sink1 space:1
+    space:1 Source space:1 Replicate space:3
+    space:5 Sink2 space:1
+
+    Source-- "input" -->Replicate
+    Replicate-- "output" -->Sink1
+    Replicate-- "output" -->Sink2
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Replicate block
+    class Source1 hidden
+    class Source2 hidden
+    class Sink1 hidden
+    class Sink2 hidden

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

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

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

@@ -0,0 +1,11 @@
+block-beta
+    columns 7
+    Source space:2 Sort space:2 Sink
+    Source-- "input" -->Sort
+    Sort-- "output" -->Sink
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Sort block
+    class Source hidden
+    class Sink hidden

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

@@ -0,0 +1,22 @@
+sequenceDiagram
+    autonumber
+    participant BlockA as Another block
+    participant Sort.input as Sort.input port
+    participant Sort as Sort block
+    participant Sort.output as Sort.output port
+    participant BlockB as Another block
+
+    BlockA-->>Sort: Connect
+    Sort-->>BlockB: Connect
+
+    loop Sort process
+        BlockA->>Sort: Message
+        Sort->>Sort: Store message
+    end
+
+    Sort->>Sort: Sort messages
+    Sort->>BlockB: Message
+    BlockA-->>Sort: Disconnect
+    Sort-->>Sort.input: Close
+    Sort-->>Sort.output: Close
+    Sort-->>BlockB: Disconnect

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")]