Implement some flow blocks

README.md

@@ -112,7 +112,10 @@ The built-in blocks provided by Protoflow are listed below:
 
 | Block             | Description                                                                                                                    |
 |:------------------|:-------------------------------------------------------------------------------------------------------------------------------|
+| [`Batch`]         | Batches input strem into chunks of a specified size.                                                                           |
 | [`Buffer`]        | Stores all messages it receives.                                                                                               |
+| [`Concat`]        | Concatenates multiple input message streams into a single output stream.                                                       |
+| [`Concat`]        | Concatenates multiple input message streams into a single output stream.                                                       |
 | [`ConcatStrings`] | Concatenates the received string messages, with an optional delimiter string inserted between each message.                    |
 | [`Const`]         | Sends a constant value.                                                                                                        |
 | [`Count`]         | Counts the number of messages it receives, while optionally passing them through.                                              |
@@ -121,24 +124,51 @@ The built-in blocks provided by Protoflow are listed below:
 | [`DecodeHex`]     | Decodes hexadecimal stream to byte stream.                                                                                     |
 | [`DecodeJSON`]    | Decodes JSON messages from a byte stream.                                                                                      |
 | [`Delay`]         | Passes messages through while delaying them by a fixed or random duration.                                                     |
+| [`Distinct`]      | Removes duplicate values from the input stream.                                                                                |
 | [`Drop`]          | Discards all messages it receives.                                                                                             |
 | [`Encode`]        | Encodes messages to a byte stream.                                                                                             |
 | [`EncodeCSV`]     | Encodes the provided header and rows, given as `prost_types::Value`, into a CSV-formatted byte stream.                         |
 | [`EncodeHex`]     | Encodes a byte stream into hexadecimal form.                                                                                   |
 | [`EncodeJSON`]    | Encodes messages into JSON format.                                                                                             |
 | [`Hash`]          | Computes the cryptographic hash of a byte stream.                                                                              |
+| [`Merge`]         | Merges multiple input message streams into a single output stream by interleaving messages as they arrive.                   |
 | [`Random`]        | Generates and sends a random value.                                                                                            |
 | [`ReadDir`]       | Reads file names from a file system directory.                                                                                 |
 | [`ReadEnv`]       | Reads the value of an environment variable.                                                                                    |
 | [`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                                                                                                   |
 | [`WriteStderr`]   | Writes bytes to standard error (aka stderr).                                                                                   |
 | [`WriteStdout`]   | Writes bytes to standard output (aka stdout).                                                                                  |
 
+#### [`Batch`]
+
+A block that simply stores all messages it receives.
+
+```mermaid
+block-beta
+    columns 7
+    Source space:2 Batch space:2 Sink
+    Source-- "input" -->Batch
+    Batch-- "output" -->Sink
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Batch block
+    class Source hidden
+    class Sink hidden
+```
+
+```bash
+protoflow execute Batch
+```
+
 #### [`Buffer`]
 
 A block that simply stores all messages it receives.
@@ -159,6 +189,32 @@ block-beta
 protoflow execute Buffer
 ```
 
+#### [`Concat`]
+
+Concatenates multiple input message streams into a single output stream.
+
+```mermaid
+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
+```
+
+```bash
+protoflow execute Concat
+```
+
 #### [`ConcatStrings`]
 
 A block for concatenating all string messages it receives, with an optional delimiter string inserted between each message
@@ -250,6 +306,28 @@ block-beta
 protoflow execute Decode encoding=text
 ```
 
+#### [`Distinct`]
+
+Removes duplicate values from the input stream.
+
+```mermaid
+block-beta
+    columns 7
+    Source space:2 Distinct space:2 Sink
+    Source-- "input" -->Distinct
+    Distinct-- "output" -->Sink
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Distinct block
+    class Source hidden
+    class Sink hidden
+```
+
+```bash
+protoflow execute Distinct
+```
+
 #### [`DecodeCSV`]
 
 A block that decodes CSV files from a byte stream into a header and rows represented as `prost_types::Value`
@@ -483,6 +561,32 @@ block-beta
 protoflow execute Hash algorithm=blake3
 ```
 
+#### [`Merge`]
+
+Combines multiple input message streams into a single output stream by interleaving messages as they arrive.
+
+```mermaid
+block-beta
+    columns 7
+    space:1 Source1 space:5
+    space:3 Merge space:1 Sink space:1
+    space:1 Source2 space:5
+    Source1-- "input" -->Merge
+    Source2-- "input" -->Merge
+    Merge-- "output" -->Sink
+
+    classDef block height:48px,padding:8px;
+    classDef hidden visibility:none;
+    class Merge block
+    class Source1 hidden
+    class Source2 hidden
+    class Sink hidden
+```
+
+```bash
+protoflow execute Merge
+```
+
 #### [`Random`]
 
 A block for generating and sending a random value.
@@ -618,6 +722,83 @@ block-beta
 protoflow execute ReadStdin < input.txt
 ```
 
+#### [`Replicate`]
+
+Duplicates a single input message stream into multiple identical output streams.
+
+```mermaid
+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
+```
+
+```bash
+protoflow execute Replicate
+```
+
+#### [`Sort`]
+
+Sorts a single input message stream in ascending order.
+
+```mermaid
+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
+    class Sink2 hidden
+```
+
+```bash
+protoflow execute Sort
+```
+
+#### [`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
@@ -794,7 +975,9 @@ To add a new block type implementation, make sure to examine and amend:
 [`echo_lines`]: lib/protoflow/examples/echo_lines
 [`examples`]: lib/protoflow/examples
 
+[`Batch`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Batch.html
 [`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
@@ -803,18 +986,23 @@ To add a new block type implementation, make sure to examine and amend:
 [`DecodeHex`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.DecodeHex.html
 [`DecodeJSON`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.DecodeJson.html
 [`Delay`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Delay.html
+[`Distinct`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Distinct.html
 [`Drop`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Drop.html
 [`Encode`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Encode.html
 [`EncodeCSV`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.EncodeCsv.html
 [`EncodeHex`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.EncodeHex.html
 [`EncodeJSON`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.EncodeJson.html
 [`Hash`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Hash.html
+[`Merge`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Merge.html
 [`Random`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.Random.html
 [`ReadDir`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.ReadDir.html
 [`ReadEnv`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.ReadEnv.html
 [`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/batch.mmd

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

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

@@ -0,0 +1,26 @@
+sequenceDiagram
+    autonumber
+    participant BlockA as Another block
+    participant Batch.input as Batch.input port
+    participant Batch as Batch block
+    participant Batch.output as Batch.output port
+    participant BlockB as Another block
+
+    BlockA-->>Batch: Connect
+    Batch-->>BlockB: Connect
+
+    loop Batch process
+        loop loop Until batch size is reached
+            BlockA->>Batch: Message
+            Batch->>Batch: Store Batch messages
+        end
+        loop Send buffered messages
+            Batch->>BlockB: Message
+        end
+    end
+
+
+    BlockA-->>Batch: Disconnect
+    Batch-->>Batch.input: Close
+    Batch-->>Batch.output: Close
+    Batch-->>BlockB: Disconnect

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

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

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

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