Datum

Appearance

Context Propagation ​

Context propagation keeps metadata such as offsets, trace IDs, tenant IDs, and commit handles next to the stream element they describe. Datum mirrors Akka's SourceWithContext and FlowWithContext wrappers by representing each element as a (value, context) pair and exposing operators that preserve the one-to-one relationship.

SourceWithContext ​

Source::as_source_with_context(extract) converts a plain Source<Out, Mat> into a SourceWithContext<Out, Ctx, Mat>. The extractor receives &Out and builds the context for that element.

SourceWithContext exposes only operators that keep every element glued to its context: map, filter / filter_not, filter_map, map_concat (duplicates the context across the expansion), map_context, map_async, grouped / sliding (bundle data and context into parallel Vecs), and via. Terminate with to / to_mat, or collect eagerly with run_collect. Convert back to a plain source with as_source() when a downstream API expects Source<(Out, Ctx), Mat>.

rust
use datum::{FlowWithContext, Sink, Source};

#[derive(Clone, Debug, PartialEq, Eq)]
struct Message {
    offset: u64,
    trace_id: String,
    payload: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct RawContext {
    offset: u64,
    trace_id: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct CommitContext {
    offset: u64,
    trace_id: String,
}

let messages = vec![
    Message {
        offset: 41,
        trace_id: "trace-a".to_owned(),
        payload: "alpha".to_owned(),
    },
    Message {
        offset: 42,
        trace_id: "trace-b".to_owned(),
        payload: "bravo".to_owned(),
    },
];

let processed: Vec<(String, CommitContext)> = Source::from_iter(messages)
    .as_source_with_context(|message| RawContext {
        offset: message.offset,
        trace_id: message.trace_id.clone(),
    })
    .map(|message| message.payload.to_uppercase())
    .via(
        FlowWithContext::<String, RawContext, String, RawContext>::identity()
            .map(|payload| format!("processed:{payload}")),
    )
    .map_context(|context| CommitContext {
        offset: context.offset,
        trace_id: context.trace_id,
    })
    .as_source()
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

The example transforms only the payload while carrying the original offset and trace ID to the commit-facing output.

FlowWithContext ​

Flow::as_flow_with_context(collapse, extract) adapts a plain flow into a FlowWithContext<In, CtxIn, Out, CtxOut, Mat>.

  • collapse(value, context) builds the plain input consumed by the wrapped flow.
  • extract(&out) builds the output context after the wrapped flow emits.

FlowWithContext exposes the same context-preserving operator set as SourceWithContext — map, filter / filter_not, filter_map, map_concat, map_context, map_async, grouped / sliding, and via — plus the to / to_mat terminals. Build an identity flow with FlowWithContext::identity(), and convert back to a plain Flow<(In, CtxIn), (Out, CtxOut), Mat> with as_flow().

rust
use datum::{Flow, NotUsed, Sink, Source};

#[derive(Clone, Debug, PartialEq, Eq)]
struct Work {
    offset: u64,
    payload: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct Context {
    offset: u64,
}

let normalize = Flow::<Work, Work, NotUsed>::identity()
    .map(|work| Work {
        offset: work.offset,
        payload: work.payload.trim().to_owned(),
    })
    .as_flow_with_context(
        |payload: String, context: Context| Work {
            offset: context.offset,
            payload,
        },
        |work| Context {
            offset: work.offset,
        },
    )
    .map(|work| work.payload)
    .as_flow();

let output: Vec<(String, Context)> =
    Source::from_iter(vec![("  payload  ".to_owned(), Context { offset: 7 })])
        .via(normalize)
        .run_with(Sink::collect())
        .unwrap()
        .wait()
        .unwrap();

Actor ask with context ​

ActorFlow::ask_with_context sends only the element to the actor message builder, keeps the context in the stream runtime, and reattaches it to the actor reply. The status variant, ActorFlow::ask_with_status_and_context, unwraps ActorStatus::Ok(value) or fails the stream for ActorStatus::Err(error) while preserving the same context behavior.

rust
use datum::actor::{Actor, ActorProcessingErr, ActorRef};
use datum::{ActorFlow, ActorStatus, ReplyPort, Source};
use std::time::Duration;

enum EnrichMsg {
    Plain {
        value: u64,
        reply_to: ReplyPort<String>,
    },
    Status {
        value: u64,
        reply_to: ReplyPort<ActorStatus<String>>,
    },
}

struct EnrichActor;

impl Actor for EnrichActor {
    type Msg = EnrichMsg;
    type State = ();
    type Arguments = ();

    async fn pre_start(
        &self,
        _myself: ActorRef<Self::Msg>,
        _args: Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        Ok(())
    }

    async fn handle(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        _state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        match message {
            EnrichMsg::Plain { value, reply_to } => {
                let _ = reply_to.send(format!("enriched-{value}"));
            }
            EnrichMsg::Status { value, reply_to } => {
                let _ = reply_to.send(ActorStatus::Ok(format!("status-{value}")));
            }
        }
        Ok(())
    }
}

let rt = tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .build()
    .unwrap();
let (actor_ref, _handle) = rt.block_on(async {
    Actor::spawn(None, EnrichActor, ())
        .await
        .expect("actor spawns")
});

let plain = Source::from_iter(1_u64..=2)
    .as_source_with_context(|value| format!("trace-{value}"))
    .via(ActorFlow::ask_with_context(
        actor_ref.clone(),
        2,
        Duration::from_secs(1),
        |value, reply_to| EnrichMsg::Plain { value, reply_to },
    ))
    .run_collect()
    .unwrap();

let status = Source::single(3_u64)
    .as_source_with_context(|value| format!("trace-{value}"))
    .via(ActorFlow::ask_with_status_and_context(
        actor_ref.clone(),
        1,
        Duration::from_secs(1),
        |value, reply_to| EnrichMsg::Status { value, reply_to },
    ))
    .run_collect()
    .unwrap();

actor_ref.stop(None);

Both variants preserve input order, matching ActorFlow::ask.

When to use context wrappers ​

Use context wrappers when metadata must stay aligned with exactly one element: offsets that must be committed after processing, trace IDs that must accompany logs or actor replies, or routing metadata that is needed after enrichment.

Avoid arbitrary dropping, reordering, or expanding when the context has one-to-one semantics. The wrapper API intentionally exposes a smaller operator set than plain Source and Flow so context cannot be detached accidentally.

Next steps ​