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/// ```
36#[derive(Default)]
37pub struct ContextOutboundInterceptor {
38    error_policy: ErrorPolicy,
39}
40
41impl ContextOutboundInterceptor {
42    /// Create with a specific error policy.
43    pub fn new(error_policy: ErrorPolicy) -> Self {
44        Self { error_policy }
45    }
46}
47
48impl OutboundInterceptor for ContextOutboundInterceptor {
49    fn name(&self) -> &'static str {
50        "dcontext-outbound"
51    }
52
53    fn on_send(
54        &self,
55        ctx: &OutboundContext<'_>,
56        _runtime_headers: &RuntimeHeaders,
57        headers: &mut Headers,
58        _message: &dyn Any,
59    ) -> Disposition {
60        if ctx.remote {
61            // Remote target: serialize context to wire bytes.
62            match dcontext::capture().serialize() {
63                Ok(bytes) => {
64                    headers.insert(ContextHeader { bytes });
65                }
66                Err(e) => {
67                    tracing::warn!(
68                        target: "dcontext_dactor",
69                        error = %e,
70                        target_actor = %ctx.target_name,
71                        "failed to serialize dcontext for outbound message"
72                    );
73                    if self.error_policy == ErrorPolicy::Reject {
74                        return Disposition::Reject(format!("dcontext serialization failed: {e}"));
75                    }
76                }
77            }
78        } else {
79            // Local target: snapshot only — no serialization needed.
80            let snap = dcontext::capture();
81            headers.insert(ContextSnapshotHeader { snapshot: snap });
82        }
83
84        Disposition::Continue
85    }
86
87    fn on_reply(
88        &self,
89        _ctx: &OutboundContext<'_>,
90        _runtime_headers: &RuntimeHeaders,
91        _headers: &Headers,
92        _outcome: &Outcome<'_>,
93    ) {
94        // No action on reply — context flows forward, not backward.
95    }
96}
dcontext_dactor/outbound.rs

dcontext_dactor/
outbound.rs
