Skip to content
This repository was archived by the owner on Jan 22, 2026. It is now read-only.

Add MapFrom block #35

Closed
imunproductive wants to merge 11 commits into from
Closed

Add MapFrom block #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
5d36817
add Mapper block
imunproductive Dec 17, 2024
629cc3c
add Mapper block to README
imunproductive Dec 17, 2024
6cd9ee5
fix tests
imunproductive Dec 17, 2024
9738f1b
fix warnings
imunproductive Dec 18, 2024
abb9519
fix mermaid
imunproductive Dec 18, 2024
a1f5c09
Merge branch 'master' of https://github.com/asimov-platform/protoflow…
imunproductive Dec 18, 2024
3db3bad
rename Mapper to MapFrom
imunproductive Dec 26, 2024
48877f1
add missing implementation
imunproductive Dec 26, 2024
87c7d60
implement stdio
imunproductive Dec 26, 2024
e21a023
Merge branch 'master' into feat/mapping
artob Jan 22, 2026
491e617
Merge branch 'master' into feat/mapping
artob Jan 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ The built-in blocks provided by Protoflow are listed below:
| [`EncodeHex`] | Encodes a byte stream into hexadecimal form. |
| [`EncodeJSON`] | Encodes messages into JSON format. |
| [`Hash`] | Computes the cryptographic hash of a byte stream. |
| [`MapFrom`] | Maps a message from one type to another. |
| [`Random`] | Generates and sends a random value. |
| [`ReadDir`] | Reads file names from a file system directory. |
| [`ReadEnv`] | Reads the value of an environment variable. |
Expand Down Expand Up @@ -483,6 +484,24 @@ block-beta
protoflow execute Hash algorithm=blake3
```

#### [`MapFrom`]

A block to map a message from one type to another.

```mermaid
block-beta
columns 7
Source space:2 MapFrom space:2 Sink
Source-- "input" -->MapFrom
MapFrom-- "output" -->Sink

classDef block height:48px,padding:8px;
classDef hidden visibility:none;
class MapFrom block
class Source hidden
class Sink hidden
```

#### [`Random`]

A block for generating and sending a random value.
Expand Down Expand Up @@ -809,6 +828,7 @@ To add a new block type implementation, make sure to examine and amend:
[`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
[`MapFrom`]: https://docs.rs/protoflow-blocks/latest/protoflow_blocks/struct.MapFrom.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
Expand Down
11 changes: 11 additions & 0 deletions lib/protoflow-blocks/doc/core/map_from.mmd
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
block-beta
columns 7
Source space:2 MapFrom space:2 Sink
Source-- "input" -->MapFrom
MapFrom-- "output" -->Sink

classDef block height:48px,padding:8px;
classDef hidden visibility:none;
class MapFrom block
class Source hidden
class Sink hidden
21 changes: 21 additions & 0 deletions lib/protoflow-blocks/doc/core/map_from.seq.mmd
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
sequenceDiagram
autonumber
participant BlockA as Another block
participant MapFrom.input as MapFrom.input port
participant MapFrom as MapFrom block
participant MapFrom.output as MapFrom.output port
participant BlockB as Another block

BlockA-->>MapFrom: Connect
MapFrom-->>BlockB: Connect

loop MapFrom process
BlockA->>MapFrom: Message
MapFrom->>MapFrom: Transform
MapFrom->>BlockB: Message
end

BlockA-->>MapFrom: Disconnect
MapFrom-->>MapFrom.input: Close
MapFrom-->>MapFrom.output: Close
MapFrom-->>BlockB: Disconnect
2 changes: 1 addition & 1 deletion lib/protoflow-blocks/src/block_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ impl<'de> serde::Deserialize<'de> for BlockConfig {
tag,
value: Value::Mapping(_mapping),
} => Ok(match tag.string.as_str() {
"Buffer" | "Const" | "Count" | "Delay" | "Drop" | "Random" => {
"Buffer" | "Const" | "Count" | "Delay" | "Drop" | "MapFrom" | "Random" => {
CoreBlockConfig::deserialize(value.clone())
.map(BlockConfig::Core)
.unwrap()
Expand Down
4 changes: 4 additions & 0 deletions lib/protoflow-blocks/src/block_tag.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ pub enum BlockTag {
Count,
Delay,
Drop,
MapFrom,
Random,
// FlowBlocks
// HashBlocks
Expand Down Expand Up @@ -77,6 +78,7 @@ impl BlockTag {
Count => "Count",
Delay => "Delay",
Drop => "Drop",
MapFrom => "MapFrom",
Random => "Random",
#[cfg(any(
feature = "hash-blake3",
Expand Down Expand Up @@ -128,6 +130,7 @@ impl FromStr for BlockTag {
"Count" => Count,
"Delay" => Delay,
"Drop" => Drop,
"MapFrom" => MapFrom,
"Random" => Random,
#[cfg(any(
feature = "hash-blake3",
Expand Down Expand Up @@ -204,6 +207,7 @@ impl BlockInstantiation for BlockTag {
Encode => Box::new(super::Encode::<String>::with_system(system, None)),
EncodeHex => Box::new(super::EncodeHex::with_system(system)),
EncodeJson => Box::new(super::EncodeJson::with_system(system)),
MapFrom => Box::new(super::MapFrom::<Any, Any>::with_system(system)),
#[cfg(feature = "std")]
ReadDir => Box::new(super::ReadDir::with_system(system)),
#[cfg(feature = "std")]
Expand Down
19 changes: 19 additions & 0 deletions lib/protoflow-blocks/src/blocks/core.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,10 @@ pub mod core {

fn drop<T: Message + 'static>(&mut self) -> Drop<T>;

fn map_from<Input: Message + 'static, Output: Message + From<Input> + 'static>(
&mut self,
) -> MapFrom<Input, Output>;

fn random<T: Message + 'static>(&mut self) -> Random<T>;

fn random_seeded<T: Message + 'static>(&mut self, seed: Option<u64>) -> Random<T>;
Expand All @@ -47,6 +51,7 @@ pub mod core {
Count,
Delay,
Drop,
MapFrom,
Random,
}

Expand Down Expand Up @@ -78,6 +83,11 @@ pub mod core {
input: InputPortName,
},

MapFrom {
input: InputPortName,
output: OutputPortName,
},

Random {
output: OutputPortName,
seed: Option<u64>,
Expand All @@ -93,6 +103,7 @@ pub mod core {
Count { .. } => "Count",
Delay { .. } => "Delay",
Drop { .. } => "Drop",
MapFrom { .. } => "MapFrom",
Random { .. } => "Random",
})
}
Expand All @@ -109,6 +120,7 @@ pub mod core {
}
Delay { output, .. } => vec![("output", Some(output.clone()))],
Drop { .. } => vec![],
MapFrom { output, .. } => vec![("output", Some(output.clone()))],
Random { output, .. } => vec![("output", Some(output.clone()))],
}
}
Expand All @@ -135,6 +147,10 @@ pub mod core {
// TODO: Delay::with_system(system, Some(delay.clone())))
}
Drop { .. } => Box::new(super::Drop::new(system.input_any())), // TODO: Drop::with_system(system)
MapFrom { .. } => Box::new(super::MapFrom::with_params(
system.input_any(),
system.output_any(),
)),
Random { seed, .. } => {
Box::new(super::Random::with_params(system.output::<u64>(), *seed))
// TODO: Random::with_system(system, *seed))
Expand All @@ -158,6 +174,9 @@ pub mod core {
mod drop;
pub use drop::*;

mod map_from;
pub use map_from::*;

mod random;
pub use random::*;
}
Expand Down
97 changes: 97 additions & 0 deletions lib/protoflow-blocks/src/blocks/core/map_from.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
// This is free and unencumbered software released into the public domain.

use crate::{prelude::Bytes, StdioConfig, StdioError, StdioSystem, System};
use protoflow_core::{Block, BlockResult, BlockRuntime, InputPort, Message, OutputPort};
use protoflow_derive::Block;
use simple_mermaid::mermaid;

/// A block to map a message from one type to another.
///
/// # Block Diagram
#[doc = mermaid!("../../../doc/core/map_from.mmd")]
///
/// # Sequence Diagram
#[doc = mermaid!("../../../doc/core/map_from.seq.mmd" framed)]
///
/// # Examples
///
/// ## Using the block in a system
///
/// ```rust
/// # use protoflow_blocks::*;
/// # fn main() {
/// System::build(|s| {
/// // TODO
/// });
/// # }
/// ```
///
#[derive(Block, Clone)]
pub struct MapFrom<Input: Message, Output: Message + From<Input>> {
/// The input message stream.
#[input]
pub input: InputPort<Input>,

/// The output message stream.
#[output]
pub output: OutputPort<Output>,
}

impl<Input: Message, Output: Message + From<Input>> MapFrom<Input, Output> {
pub fn new(input: InputPort<Input>, output: OutputPort<Output>) -> Self {
Self::with_params(input, output)
}
}

impl<Input: Message, Output: Message + From<Input>> MapFrom<Input, Output> {
pub fn with_params(input: InputPort<Input>, output: OutputPort<Output>) -> Self {
Self { input, output }
}
}

impl<Input: Message + 'static, Output: Message + From<Input> + 'static> MapFrom<Input, Output> {
pub fn with_system(system: &System) -> Self {
use crate::SystemBuilding;
Self::with_params(system.input(), system.output())
}
}

impl<Input: Message, Output: Message + From<Input>> Block for MapFrom<Input, Output> {
fn execute(&mut self, _: &dyn BlockRuntime) -> BlockResult {
while let Some(input) = self.input.recv()? {
let output: Output = From::from(input);
self.output.send(&output)?;
}

Ok(())
}
}

#[cfg(feature = "std")]
impl<Input: Message, Output: Message + From<Input>> StdioSystem for MapFrom<Input, Output> {
fn build_system(config: StdioConfig) -> Result<System, StdioError> {
use crate::SystemBuilding;

config.reject_any()?;

Ok(System::build(|s| {
let stdin = config.read_stdin(s);
let map = s.block(MapFrom::<Bytes, Bytes>::with_system(s));
s.connect(&stdin.output, &map.input);
}))
}
}

#[cfg(test)]
mod tests {
use super::MapFrom;
use crate::{System, SystemBuilding};

#[test]
fn instantiate_block() {
// Check that the block is constructible:
let _ = System::build(|s| {
let _ = s.block(MapFrom::<u32, u64>::with_params(s.input(), s.output()));
});
}
}
1 change: 1 addition & 0 deletions lib/protoflow-blocks/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ pub fn build_stdio_system(
"Count" => Count::<String>::build_system(config)?,
"Delay" => Delay::<String>::build_system(config)?,
"Drop" => Drop::<String>::build_system(config)?,
"MapFrom" => MapFrom::<String, String>::build_system(config)?,
"Random" => Random::<u64>::build_system(config)?,
// FlowBlocks
// HashBlocks
Expand Down
10 changes: 8 additions & 2 deletions lib/protoflow-blocks/src/system.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@ use crate::{
types::{DelayType, Encoding},
AllBlocks, Buffer, ConcatStrings, Const, CoreBlocks, Count, Decode, DecodeCsv, DecodeHex,
DecodeJson, Delay, Drop, Encode, EncodeCsv, EncodeHex, EncodeJson, FlowBlocks, HashBlocks,
IoBlocks, MathBlocks, Random, ReadDir, ReadEnv, ReadFile, ReadStdin, SplitString, SysBlocks,
TextBlocks, WriteFile, WriteStderr, WriteStdout,
IoBlocks, MapFrom, MathBlocks, Random, ReadDir, ReadEnv, ReadFile, ReadStdin, SplitString,
SysBlocks, TextBlocks, WriteFile, WriteStderr, WriteStdout,
};
#[cfg(all(feature = "std", feature = "serde"))]
use crate::{ReadSocket, WriteSocket};
Expand Down Expand Up @@ -169,6 +169,12 @@ impl CoreBlocks for System {
self.0.block(Drop::<T>::with_system(self))
}

fn map_from<Input: Message + 'static, Output: Message + From<Input> + 'static>(
&mut self,
) -> MapFrom<Input, Output> {
self.0.block(MapFrom::<Input, Output>::with_system(self))
}

fn random<T: Message + 'static>(&mut self) -> Random<T> {
self.0.block(Random::<T>::with_system(self, None))
}
Expand Down
Loading