dcontext_dactor/
outbound.rs

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

dcontext_dactor/
outbound.rs
