dcontext_dactor/
outbound.rs

1use std::any::Any;
2
3use dactor::{Disposition, Headers, OutboundContext, OutboundInterceptor, Outcome, RuntimeHeaders};
4
5use crate::header::{ContextHeader, ContextSnapshotHeader};
6use crate::ErrorPolicy;
7
8/// Outbound interceptor that captures the current dcontext and attaches it
9/// to outgoing actor messages as headers.
10///
11/// - **Local targets**: attaches a [`ContextSnapshotHeader`] only — no
12///   serialization is performed, preserving local-only context values.
13/// - **Remote targets**: serializes context to bytes and attaches a
14///   [`ContextHeader`] for wire transport.
15///
16/// # Error Handling
17///
18/// Controlled by [`ErrorPolicy`]:
19/// - [`LogAndContinue`](ErrorPolicy::LogAndContinue) (default) — log and
20///   send the message without context.
21/// - [`Reject`](ErrorPolicy::Reject) — reject the message.
22///
23/// # Usage
24///
25/// ```ignore
26/// use dcontext_dactor::{ContextOutboundInterceptor, ErrorPolicy};
27///
28/// // Default: log and continue on errors
29/// runtime.add_outbound_interceptor(Box::new(ContextOutboundInterceptor::default()));
30///
31/// // Strict: reject messages if context serialization fails
32/// runtime.add_outbound_interceptor(Box::new(
33///     ContextOutboundInterceptor::new(ErrorPolicy::Reject),
34/// ));
35/// ```
36pub struct ContextOutboundInterceptor {
37    error_policy: ErrorPolicy,
38}
39
40impl ContextOutboundInterceptor {
41    /// Create with a specific error policy.
42    pub fn new(error_policy: ErrorPolicy) -> Self {
43        Self { error_policy }
44    }
45}
46
47impl Default for ContextOutboundInterceptor {
48    fn default() -> Self {
49        Self {
50            error_policy: ErrorPolicy::default(),
51        }
52    }
53}
54
55impl OutboundInterceptor for ContextOutboundInterceptor {
56    fn name(&self) -> &'static str {
57        "dcontext-outbound"
58    }
59
60    fn on_send(
61        &self,
62        ctx: &OutboundContext<'_>,
63        _runtime_headers: &RuntimeHeaders,
64        headers: &mut Headers,
65        _message: &dyn Any,
66    ) -> Disposition {
67        if ctx.remote {
68            // Remote target: serialize context to wire bytes.
69            match dcontext::async_ctx::serialize_context() {
70                Ok(bytes) => {
71                    headers.insert(ContextHeader { bytes });
72                }
73                Err(e) => {
74                    tracing::warn!(
75                        target: "dcontext_dactor",
76                        error = %e,
77                        target_actor = %ctx.target_name,
78                        "failed to serialize dcontext for outbound message"
79                    );
80                    if self.error_policy == ErrorPolicy::Reject {
81                        return Disposition::Reject(format!("dcontext serialization failed: {e}"));
82                    }
83                }
84            }
85        } else {
86            // Local target: snapshot only — no serialization needed.
87            let snap = dcontext::async_ctx::snapshot();
88            headers.insert(ContextSnapshotHeader { snapshot: snap });
89        }
90
91        Disposition::Continue
92    }
93
94    fn on_reply(
95        &self,
96        _ctx: &OutboundContext<'_>,
97        _runtime_headers: &RuntimeHeaders,
98        _headers: &Headers,
99        _outcome: &Outcome<'_>,
100    ) {
101        // No action on reply — context flows forward, not backward.
102    }
103}
dcontext_dactor/outbound.rs

dcontext_dactor/
outbound.rs
