Datum

Appearance

Modularity And Composition ​

Datum stream pieces are ordinary immutable blueprints. You can build reusable Flows, compose them into larger Source -> Flow -> Sink pipelines, keep or transform materialized values, and package graph fragments for reuse inside GraphDsl.

Reusable flows ​

A reusable flow is usually just a function that returns Flow<In, Out, Mat>. Attach it to a source with Source::via(flow) or Source::via_mat(flow, combine). Compose two flows with Flow::via or Flow::via_mat.

Use the *_mat forms when both sides have materialized values you care about. Keep::left, Keep::right, Keep::both, and Keep::none cover the common cases; a custom closure can build any combined value. Use map_materialized_value when a reusable component should hide, replace, or wrap the materialized value it gets from its internals.

rust
use datum::{Attributes, Flow, Keep, Sink, Source};

fn tag_non_empty(prefix: &'static str) -> Flow<&'static str, String, &'static str> {
    Flow::identity()
        .map(|item: &'static str| item.trim().to_ascii_lowercase())
        .filter(|item| !item.is_empty())
        .map(move |item| format!("{prefix}:{item}"))
        .named("tag-non-empty")
        .map_materialized_value(move |_| prefix)
}

let suffix = Flow::identity()
    .map(|item: String| format!("{item}!"))
    .with_attributes(Attributes::named("suffix"))
    .map_materialized_value(|_| "suffix");

let ((tag_mat, suffix_mat), completion) = Source::from_iter(["  A  ", "", " B "])
    .via_mat(tag_non_empty("order"), Keep::right)
    .via_mat(suffix, Keep::both)
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

let values = completion.wait().unwrap();

Source::to(sink) builds a RunnableGraph and keeps the source materialized value. Source::to_mat(sink, combine) lets you choose how source and sink materialized values are combined. The same pattern exists at the flow boundary: Flow::to(sink) and Flow::to_mat(sink, combine) build a Sink.

BidiFlow ​

BidiFlow<I1, O1, I2, O2> packages two directional flows: a top flow from I1 to O1, and a bottom flow from I2 to O2. This mirrors Akka's bidirectional flow shape and is useful for protocol stacks, codecs, framing, and symmetric adapters.

  • BidiFlow::from_flows(top, bottom) builds a bidirectional flow from two ordinary flows.
  • join(flow) places an ordinary Flow<O1, I2, _> between the top and bottom sides and returns Flow<I1, O2, NotUsed>.
  • atop(other) stacks two bidirectional layers.
  • reversed() swaps the top and bottom sides.
rust
use datum::{BidiFlow, Flow, Source};

let codec = BidiFlow::from_flows(
    Flow::identity().map(|message: String| format!("wire:{message}")),
    Flow::identity().map(|wire: String| wire.trim_start_matches("wire:").to_owned()),
)
.named("codec");

let framing = BidiFlow::from_flows(
    Flow::identity().map(|wire: String| format!("<{wire}>")),
    Flow::identity().map(|frame: String| {
        frame
            .trim_start_matches('<')
            .trim_end_matches('>')
            .to_owned()
    }),
)
.named("framing");

let protocol = codec.clone().atop(framing).named("client-protocol");
let loopback = Flow::identity().map(|frame: String| frame.replace("ping", "pong"));

let responses = Source::single("ping".to_owned())
    .via(protocol.join(loopback))
    .run_collect()
    .unwrap();

let reversed_round_trip = Source::single("wire:ack".to_owned())
    .via(codec.reversed().join(Flow::identity()))
    .run_collect()
    .unwrap();

In the example, outbound messages are encoded and framed, the local transport flow changes the wire frame, and the inbound side unframes and decodes the response.

Partial graph fragments ​

For reusable graph topology, use GraphDsl::partial. It returns a PartialGraph<S> whose shape can be imported into a parent graph with GraphBuilder::import(&fragment).

rust
use datum::{GraphDsl, GraphFlowShape, MapStage, PartialGraph};

let fragment: PartialGraph<GraphFlowShape<u64, u64>> = GraphDsl::partial(|builder| {
    let plus_one = builder.add(MapStage::new(|item: u64| item + 1));
    let times_two = builder.add(MapStage::new(|item: u64| item * 2));

    builder.connect(plus_one.outlet(), times_two.inlet())?;

    Ok(GraphFlowShape::new(plus_one.inlet(), times_two.outlet()))
})
.named("increment-then-double");

let graph = GraphDsl::try_create(|builder| {
    let imported = builder.import(&fragment)?;
    let add_ten = builder.add(MapStage::new(|item: u64| item + 10));

    builder.connect(imported.outlet(), add_ten.inlet())?;

    Ok(GraphFlowShape::new(imported.inlet(), add_ten.outlet()))
})
.unwrap();

let values = graph.run_with_input([1_u64, 2]).unwrap();

partial is still a blueprint: it does not allocate a running stream. Its builder closure is evaluated when a parent graph imports the fragment.

Names and attributes ​

Names and attributes live on blueprints and graph stages. Use them to label stages, provide input buffer hints, or carry dispatcher names:

  • named(name) adds Attributes::named(name).
  • with_attributes(attrs) replaces the current attributes.
  • add_attributes(attrs) merges attributes with the existing set.
rust
use datum::{Attributes, Flow};

let flow: Flow<u64, u64> = Flow::identity()
    .named("parse-orders")
    .add_attributes(Attributes::input_buffer(4, 16));

assert_eq!(flow.attributes().name(), Some("parse-orders"));
assert_eq!(flow.attributes().input_buffer_hint(), Some((4, 16)));

let replaced = flow.with_attributes(Attributes::named("replacement"));

assert_eq!(replaced.attributes().name(), Some("replacement"));
assert_eq!(replaced.attributes().input_buffer_hint(), None);

The common constructors are Attributes::named, Attributes::input_buffer, and Attributes::dispatcher.

Next steps ​