dcontext_dactor/
lib.rs

1//! # dcontext-dactor
2//!
3//! Automatic dcontext propagation through [dactor](https://github.com/Yaming-Hub/dactor)
4//! actor messages.
5//!
6//! This crate bridges [`dcontext`] (distributed context propagation) with
7//! [`dactor`] (actor framework) by providing interceptors that automatically
8//! carry context across actor message boundaries — **no boilerplate needed
9//! in handlers**.
10//!
11//! ## How It Works
12//!
13//! Context propagation is a **two-stage interceptor pipeline**:
14//!
15//! 1. **Outbound interceptor** (sender side) — [`ContextOutboundInterceptor`]
16//!    captures the current dcontext and attaches it as message headers.
17//!    - Local targets → [`ContextSnapshotHeader`] (preserves local-only values)
18//!    - Remote targets → [`ContextHeader`] (serialized wire bytes)
19//!
20//! 2. **Inbound interceptor** (receiver side) — [`ContextInboundInterceptor`]
21//!    performs two actions:
22//!    - `on_receive`: normalizes headers — deserializes wire bytes into a
23//!      [`ContextSnapshotHeader`] if needed.
24//!    - `wrap_handler`: wraps the handler future with
25//!      [`dcontext::with_context`], restoring the propagated snapshot into
26//!      the async task-local scope automatically.
27//!
28//! This uses dactor 0.3's `wrap_handler` feature to wrap the handler future
29//! with a task-local context scope. `dcontext::get_context` /
30//! `dcontext::set_context` work transparently inside the handler.
31//!
32//! ## Error Handling
33//!
34//! Both interceptors accept an [`ErrorPolicy`] that controls behavior when
35//! serialization or deserialization fails:
36//!
37//! - [`ErrorPolicy::LogAndContinue`] (default) — log a warning, deliver the
38//!   message without context.
39//! - [`ErrorPolicy::Reject`] — reject the message via
40//!   [`Disposition::Reject`](dactor::Disposition::Reject).
41//!
42//! ## Local vs. Remote
43//!
44//! - **Local** (same process): Context is propagated via
45//!   [`ContextSnapshot`](dcontext::ContextSnapshot), preserving local-only
46//!   values that cannot be serialized. **No serialization is performed.**
47//!
48//! - **Remote** (cross-process): Context is serialized to bytes via
49//!   [`dcontext::serialize_context`] and transmitted as a wire header.
50//!   Local-only values are excluded.
51//!
52//! ## Quick Start
53//!
54//! ```ignore
55//! use dcontext_dactor::{ContextOutboundInterceptor, ContextInboundInterceptor};
56//!
57//! // Register interceptors on your runtime — that's it!
58//! runtime.add_outbound_interceptor(Box::new(ContextOutboundInterceptor::default()));
59//! runtime.add_inbound_interceptor(Box::new(ContextInboundInterceptor::default()));
60//!
61//! // Handlers automatically have dcontext available — no boilerplate
62//! #[async_trait]
63//! impl Handler<MyMessage> for MyActor {
64//!     async fn handle(&mut self, msg: MyMessage, ctx: &mut ActorContext) -> () {
65//!         // dcontext is automatically restored by the interceptor's wrap_handler
66//!         let rid: RequestId = dcontext::get_context("request_id");
67//!         // ... handle message with context available ...
68//!     }
69//! }
70//! ```
71//!
72//! ## Manual Extraction
73//!
74//! For advanced use cases (e.g., spawning sub-tasks that need context),
75//! [`extract_context`] is still available to pull the snapshot from headers.
76//! [`with_propagated_context`] is retained for backward compatibility but is
77//! no longer needed when using the interceptor pipeline.
78
79mod header;
80mod inbound;
81mod outbound;
82mod propagation;
83
84pub use header::{ContextHeader, ContextSnapshotHeader};
85pub use inbound::ContextInboundInterceptor;
86pub use outbound::ContextOutboundInterceptor;
87pub use propagation::extract_context;
88
89#[allow(deprecated)]
90pub use propagation::with_propagated_context;
91
92/// Register dcontext header deserializers with a dactor [`HeaderRegistry`](dactor::HeaderRegistry).
93///
94/// Call this at startup before processing remote messages so that
95/// [`ContextHeader`] can be reconstructed from wire bytes during remote
96/// actor communication.
97///
98/// # Example
99///
100/// ```ignore
101/// use dactor::HeaderRegistry;
102/// use dcontext_dactor::register_context_headers;
103///
104/// let mut registry = HeaderRegistry::new();
105/// register_context_headers(&mut registry);
106/// // pass registry to your transport / runtime
107/// ```
108pub fn register_context_headers(registry: &mut dactor::HeaderRegistry) {
109    registry.register("dcontext.wire", |bytes: &[u8]| {
110        Some(Box::new(ContextHeader::from_bytes(bytes.to_vec())))
111    });
112}
113
114/// Controls how interceptors behave when serialization or deserialization fails.
115///
116/// Passed to [`ContextOutboundInterceptor::new`] and
117/// [`ContextInboundInterceptor::new`] to configure error handling.
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
119pub enum ErrorPolicy {
120    /// Log a warning and deliver the message without propagated context.
121    /// This is the default.
122    LogAndContinue,
123    /// Reject the message via [`Disposition::Reject`](dactor::Disposition::Reject).
124    Reject,
125}
126
127impl Default for ErrorPolicy {
128    fn default() -> Self {
129        Self::LogAndContinue
130    }
131}
132
133#[cfg(test)]
134mod tests;
dcontext_dactor/lib.rs

dcontext_dactor/
lib.rs
