Actors Interop

Datum integrates with Ractor actors through three bridge APIs:

• ActorSource — an actor-backed source that lets external code push elements into a stream.
• ActorSink — routes stream elements to an actor's mailbox.
• ActorFlow::ask — transforms each stream element by sending it to an actor and waiting for a reply.
• ActorPubSub — named broadcast groups backed by Ractor's pg process groups.
• StreamRefs — one-shot SourceRef / SinkRef handles for backpressured streaming across actor boundaries.

Actor spawning is asynchronous (Ractor runs on Tokio). ActorSource handles its own spawn internally so you can start immediately. ActorSink and ActorFlow::ask require an already-running ActorRef, which you obtain by spawning the actor before materialization.

ActorSource

ActorSource::actor_ref::<T>() builds a source whose materialized value is an ActorRef<ActorSourceMessage<T>>. The source creates and owns its own actor — no external spawning is needed. Send ActorSourceMessage::Element(x) to emit elements, ActorSourceMessage::Complete to complete normally, or ActorSourceMessage::Fail(reason) to fail the stream.

use datum::{ActorSource, ActorSourceMessage, Keep, Sink};

// ActorSource::actor_ref() materializes an ActorRef that feeds the source.
// No external actor spawning is needed: the source creates and owns its actor.
let (actor_ref, completion) = ActorSource::actor_ref::<u64>()
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

actor_ref.cast(ActorSourceMessage::Element(10)).unwrap();
actor_ref.cast(ActorSourceMessage::Element(20)).unwrap();
actor_ref.cast(ActorSourceMessage::Complete).unwrap();

The source's internal actor mailbox is unbounded. A fast sender can enqueue messages faster than downstream consumes them. Use ActorSource::actor_ref_with_backpressure when the producer may outpace demand — it sends an ack to a supplied actor after each element is consumed.

ActorSource::typed is an alias for ActorSource::actor_ref with the same protocol.

ActorSink

ActorSink::typed(actor_ref) routes each stream element to actor_ref as an ActorSinkMessage::Element(x), followed by ActorSinkMessage::Complete when the stream ends (or ActorSinkMessage::Fail(error) on failure). The receiving actor must accept ActorSinkMessage<In>.

use datum::actor::{Actor, ActorProcessingErr, ActorRef};
use datum::{ActorSink, ActorSinkMessage, Source};
use std::sync::{Arc, Mutex};
use std::time::Duration;

// A minimal actor that collects ActorSinkMessage<u64> into a shared Vec.
struct CollectorActor;

impl Actor for CollectorActor {
    type Msg = ActorSinkMessage<u64>;
    type State = Arc<Mutex<Vec<u64>>>;
    type Arguments = Arc<Mutex<Vec<u64>>>;

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

    async fn handle(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        if let ActorSinkMessage::Element(x) = message {
            state.lock().unwrap().push(x);
        }
        Ok(())
    }
}

let received = Arc::new(Mutex::new(Vec::<u64>::new()));
let rt = tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .build()
    .unwrap();
let (actor_ref, _handle) = rt.block_on(async {
    Actor::spawn(None, CollectorActor, Arc::clone(&received))
        .await
        .expect("actor spawns")
});

Source::from_iter(1_u64..=3)
    .run_with(ActorSink::typed(actor_ref))
    .unwrap()
    .wait()
    .unwrap();

// The stream has completed, but actor delivery is asynchronous — wait for
// the mailbox to drain with a bounded condition-wait (never a fixed sleep).
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while received.lock().unwrap().len() < 3 && std::time::Instant::now() < deadline {
    std::thread::sleep(Duration::from_millis(2));
}

ActorSink::actor_ref is the general form: you supply three closures that build the element, complete, and failure messages from your own enum. This avoids wrapping your existing message type in ActorSinkMessage:

use datum::actor::{Actor, ActorProcessingErr, ActorRef};
use datum::{ActorSink, Source};
use std::sync::mpsc;
use std::time::Duration;

#[derive(Debug, PartialEq, Eq)]
enum MyMsg {
    Item(u64),
    Done,
    Failed(String),
}

struct RecorderActor {
    events: mpsc::Sender<MyMsg>,
}

impl Actor for RecorderActor {
    type Msg = MyMsg;
    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> {
        self.events.send(message).unwrap();
        Ok(())
    }
}

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

Source::from_iter(1_u64..=3)
    .run_with(ActorSink::actor_ref(
        actor_ref.clone(),
        MyMsg::Item,
        || MyMsg::Done,
        |err| MyMsg::Failed(err.to_string()),
    ))
    .unwrap()
    .wait()
    .unwrap();

let events = [
    rx.recv_timeout(Duration::from_secs(1)).unwrap(),
    rx.recv_timeout(Duration::from_secs(1)).unwrap(),
    rx.recv_timeout(Duration::from_secs(1)).unwrap(),
    rx.recv_timeout(Duration::from_secs(1)).unwrap(),
];

actor_ref.stop(None);
assert_eq!(
    events,
    [MyMsg::Item(1), MyMsg::Item(2), MyMsg::Item(3), MyMsg::Done,]
);

ActorSink has no backpressure signal from the destination actor: a fast upstream can saturate the actor's mailbox. Use ActorSink::actor_ref_with_backpressure (or the typed equivalent ActorSink::typed_with_backpressure) when the actor processes slowly or the element rate is unbounded — those variants send each element only after receiving an acknowledgement from the actor.

ActorFlow::ask

ActorFlow::ask(actor_ref, parallelism, timeout, make_msg) turns each stream element into a request/reply round-trip to a Ractor actor:

• actor_ref — the target actor.
• parallelism — maximum requests in flight simultaneously (must be > 0).
• timeout — per-request deadline; the stream fails with StreamError::ActorAskTimeout if a reply is not received in time.
• make_msg: Fn(In, ReplyPort<Out>) -> Msg — builds the actor message. Your actor must call reply_to.send(value) to deliver the reply.

Output order always matches input order regardless of reply order.

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

// Define a message enum that carries the reply port.
enum DoubleMsg {
    Double {
        value: u64,
        reply_to: ReplyPort<u64>,
    },
}

// Any Send + 'static type implements ractor's Message trait automatically.

// A stateless actor that doubles each incoming value.
struct DoubleActor;

impl Actor for DoubleActor {
    type Msg = DoubleMsg;
    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> {
        let DoubleMsg::Double { value, reply_to } = message;
        let _ = reply_to.send(value * 2);
        Ok(())
    }
}

// Spawn the actor on a dedicated Tokio runtime.
let rt = tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .build()
    .unwrap();
let (actor_ref, _handle) = rt.block_on(async {
    Actor::spawn(None, DoubleActor, ())
        .await
        .expect("actor spawns")
});

// Wire ActorFlow::ask into a stream: each element becomes a request/reply pair.
let result: Vec<u64> = Source::from_iter(1_u64..=4)
    .via(ActorFlow::ask(
        actor_ref.clone(),
        2, // max in-flight requests
        Duration::from_secs(1),
        |x, reply_to| DoubleMsg::Double { value: x, reply_to },
    ))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

actor_ref.stop(None);

Any Send + 'static type satisfies Ractor's Message trait automatically in Datum's supported builds, so no explicit impl Message is needed for ordinary actor messages.

Timeouts and failures

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

// An actor that holds on to the reply port and never responds.
enum SilentMsg {
    Silent { _reply_to: ReplyPort<u64> },
}

struct SilentActor;

impl Actor for SilentActor {
    type Msg = SilentMsg;
    type State = Vec<ReplyPort<u64>>; // holds ports open to prevent Drop
    type Arguments = ();

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

    async fn handle(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        let SilentMsg::Silent { _reply_to } = message;
        state.push(_reply_to); // intentionally never send
        Ok(())
    }
}

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

let timeout = Duration::from_millis(50);
let result = Source::single(1_u64)
    .via(ActorFlow::ask(
        actor_ref.clone(),
        1,
        timeout,
        |_x, reply_to| SilentMsg::Silent {
            _reply_to: reply_to,
        },
    ))
    .run_with(Sink::collect())
    .unwrap()
    .wait();

actor_ref.stop(None);

Failure cause	StreamError variant
Reply not received within timeout	ActorAskTimeout { timeout }
Actor terminated before replying	ActorTerminated
Reply port dropped without a send	ActorAskResponseDropped
make_msg closure panics	ActorAskTaskFailed { reason }

ActorFlow::ask_with_status

When your actor needs to signal stream failures — not just return values — use ActorFlow::ask_with_status. The actor replies with ActorStatus::Ok(value) to emit an element or ActorStatus::Err(error) to fail the stream:

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

enum StatusMsg {
    Process {
        value: u64,
        reply_to: ReplyPort<ActorStatus<u64>>,
    },
}

struct StatusActor;

impl Actor for StatusActor {
    type Msg = StatusMsg;
    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> {
        let StatusMsg::Process { value, reply_to } = message;
        if value == 0 {
            let _ = reply_to.send(ActorStatus::Err(StreamError::Failed(
                "zero is not allowed".to_owned(),
            )));
        } else {
            let _ = reply_to.send(ActorStatus::Ok(value * 2));
        }
        Ok(())
    }
}

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

let values: Vec<u64> = Source::from_iter(1_u64..=3)
    .via(ActorFlow::ask_with_status(
        actor_ref.clone(),
        2,
        timeout,
        |value, reply_to| StatusMsg::Process { value, reply_to },
    ))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

let failed = Source::single(0_u64)
    .via(ActorFlow::ask_with_status(
        actor_ref.clone(),
        1,
        timeout,
        |value, reply_to| StatusMsg::Process { value, reply_to },
    ))
    .run_with(Sink::collect())
    .unwrap()
    .wait();

actor_ref.stop(None);
assert_eq!(values, vec![2, 4, 6]);
assert_eq!(
    failed,
    Err(StreamError::Failed("zero is not allowed".to_owned()))
);

ActorStatus::Err(error) fails the stream with the given StreamError, identical to an upstream failure.

ActorFlow::watch

ActorFlow::watch(actor_ref) is a pass-through flow that fails the stream if actor_ref terminates while the stream is running:

use datum::actor::{Actor, ActorProcessingErr, ActorRef};
use datum::{ActorFlow, ActorSource, ActorSourceMessage, Keep, Sink, StreamError};

enum GuardMsg {
    Stop,
}

struct GuardActor;

impl Actor for GuardActor {
    type Msg = GuardMsg;
    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> {
        myself.stop(Some("guard stopped".to_owned()));
        Ok(())
    }
}

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

let (source_ref, completion) = ActorSource::actor_ref::<u64>()
    .via(ActorFlow::watch(guarded_ref.clone()))
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

guarded_ref.cast(GuardMsg::Stop).unwrap();
let result = completion.wait();
let _ = source_ref.cast(ActorSourceMessage::Complete);
rt.block_on(async {
    guarded_handle.await.expect("actor joins");
});

assert!(
    matches!(result, Err(StreamError::Failed(reason)) if reason.contains("watched actor terminated"))
);

The flow binds Ractor's monitor supervision: once the watched actor terminates, the stream fails with StreamError::Failed containing a termination message, even if upstream is idle.

StreamRefs

StreamRefs::source_ref::<T>() and StreamRefs::sink_ref::<T>() provide one-shot SourceRef / SinkRef handles for backpressured streaming across actor boundaries. Completion, failure, cancellation, and cumulative demand propagate across the ref.

In the same process the ref is a direct in-memory splice (no serialization). Across processes, datum-core owns a transport-agnostic protobuf seam (stream_ref_proto) that datum-net carries over plaintext TCP or encrypted QUIC. StreamRefs have a dedicated guide with local examples, settings, one-shot semantics, and the remote transports: StreamRefs.

ActorPubSub

ActorPubSub uses Ractor's pg process groups as a local pub/sub bus. Each group name is a string identifier shared between sources and sinks.

ActorPubSub::source::<T>(group) subscribes its materialized actor to group. Multiple sources can subscribe to the same group simultaneously.

ActorPubSub::sink::<T>(group) broadcasts each stream element to every current group member as ActorSourceMessage::Element(x), and sends ActorSourceMessage::Complete when the stream ends.

use datum::{ActorPubSub, Keep, Sink, Source};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

static GROUP_COUNTER: AtomicUsize = AtomicUsize::new(0);

// Use a unique group name per test run to avoid cross-test interference.
let group = format!(
    "docs-pubsub-{}",
    GROUP_COUNTER.fetch_add(1, Ordering::SeqCst),
);

// Source subscribes its materialized actor to the named pg group.
let (_source_ref, completion) = ActorPubSub::source::<u64>(group.clone())
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

// Wait until the source actor has joined the group.
let deadline = Instant::now() + Duration::from_secs(2);
while ractor::pg::get_members(&group).is_empty() {
    assert!(
        Instant::now() < deadline,
        "source actor did not join pg group"
    );
    std::thread::sleep(Duration::from_millis(1));
}

// Sink broadcasts to every current member of the group.
Source::from_iter([1_u64, 2])
    .run_with(ActorPubSub::sink(group))
    .unwrap()
    .wait()
    .unwrap();

Wait for the source actor to join the group before sending (the pg::get_members poll above). The sink skips members that have terminated between the group snapshot and the send, so one stale entry does not block delivery to remaining subscribers.

Performance notes

ActorFlow::ask uses a spin-then-park wait loop per in-flight request. For a single request with parallelism 1 the steady-state round-trip includes a Ractor mailbox send, the actor's async message dispatch, and the reply-port wake. Fast-completing actors (sub-µs) are caught in the spin window. Slower actors park the consumer thread and are woken by ReplyPort::send.

The ordered_sum and StreamRefs benchmarks in roadmap/benchmarks/actor-ask.md record Datum vs Akka wall-clock and CPU numbers for these paths.

StreamRefs (SourceRef/SinkRef) take the same-process direct splice on the local path — no actors and no serialization on the hot path — which is ~6.6–7.8× faster than warmed Akka in roadmap/benchmarks/actor-ask.md. The cross-process protobuf carriers are benchmarked separately against forced-remote Akka in roadmap/benchmarks/net.md (see the StreamRefs guide for the table).

Next steps

• Futures Interop — map_async, async sources and sinks
• StreamRefs — one-shot source and sink references
• Execution Model — drain thread, Tokio dispatch, spin-then-park