caelix_core/
microservice.rs

1//! Transport-neutral microservice handler metadata.
2
3use crate::{BoxFuture, Container, Injectable, ProviderDef, Result};
4use serde_json::Value;
5use std::{collections::BTreeMap, sync::Arc, time::SystemTime};
6use tokio_util::sync::CancellationToken;
7
8/// The kind of a message handler.
9#[derive(Clone, Copy, Debug, Eq, PartialEq)]
10pub enum MessageHandlerKind {
11    /// A request/reply command handler.
12    Command,
13    /// A durable, at-least-once event handler.
14    Event,
15}
16
17/// Durable transport delivery details attached to an event handler invocation.
18#[derive(Clone, Debug, Default)]
19pub struct MessageDelivery {
20    /// The stream that delivered the message, when applicable.
21    pub stream: Option<String>,
22    /// The durable consumer that delivered the message, when applicable.
23    pub consumer: Option<String>,
24    /// The one-based delivery attempt.
25    pub attempt: u64,
26}
27
28/// Context made available to a microservice handler.
29#[derive(Clone, Debug)]
30pub struct MessageContext {
31    subject: String,
32    headers: BTreeMap<String, String>,
33    correlation_id: Option<String>,
34    deadline: Option<SystemTime>,
35    cancellation: CancellationToken,
36    delivery: Option<MessageDelivery>,
37    event_id: Option<String>,
38}
39
40impl MessageContext {
41    /// Creates a context for a transport delivery.
42    pub fn new(
43        subject: impl Into<String>,
44        headers: BTreeMap<String, String>,
45        correlation_id: Option<String>,
46        deadline: Option<SystemTime>,
47        cancellation: CancellationToken,
48        delivery: Option<MessageDelivery>,
49        event_id: Option<String>,
50    ) -> Self {
51        Self {
52            subject: subject.into(),
53            headers,
54            correlation_id,
55            deadline,
56            cancellation,
57            delivery,
58            event_id,
59        }
60    }
61
62    /// The subject that delivered this message.
63    pub fn subject(&self) -> &str {
64        &self.subject
65    }
66
67    /// Application headers propagated by the transport.
68    pub fn headers(&self) -> &BTreeMap<String, String> {
69        &self.headers
70    }
71
72    /// The request correlation identifier, when this is a command.
73    pub fn correlation_id(&self) -> Option<&str> {
74        self.correlation_id.as_deref()
75    }
76
77    /// The requested deadline, when one was propagated.
78    pub fn deadline(&self) -> Option<SystemTime> {
79        self.deadline
80    }
81
82    /// A token cancelled when the application begins shutdown.
83    pub fn cancellation_token(&self) -> &CancellationToken {
84        &self.cancellation
85    }
86
87    /// Delivery metadata for durable events.
88    pub fn delivery(&self) -> Option<&MessageDelivery> {
89        self.delivery.as_ref()
90    }
91
92    /// The one-based delivery attempt, or zero for non-durable commands.
93    pub fn delivery_attempt(&self) -> u64 {
94        self.delivery
95            .as_ref()
96            .map_or(0, |delivery| delivery.attempt)
97    }
98
99    /// The stable event envelope ID, when this is a durable event delivery.
100    pub fn event_id(&self) -> Option<&str> {
101        self.event_id.as_deref()
102    }
103}
104
105type InvokeMessageFn = Arc<
106    dyn for<'a> Fn(&'a Container, MessageContext, Value) -> BoxFuture<'a, Result<Option<Value>>>
107        + Send
108        + Sync,
109>;
110
111/// One transport-neutral message handler declared by a microservice.
112#[derive(Clone)]
113pub struct MessageHandlerDef {
114    /// Whether the handler is a command or event handler.
115    pub kind: MessageHandlerKind,
116    /// The logical subject pattern.
117    pub pattern: &'static str,
118    invoke: InvokeMessageFn,
119}
120
121impl MessageHandlerDef {
122    /// Builds a handler definition. This is primarily used by `#[microservice]`.
123    pub fn new(
124        kind: MessageHandlerKind,
125        pattern: &'static str,
126        invoke: impl for<'a> Fn(
127            &'a Container,
128            MessageContext,
129            Value,
130        ) -> BoxFuture<'a, Result<Option<Value>>>
131        + Send
132        + Sync
133        + 'static,
134    ) -> Self {
135        Self {
136            kind,
137            pattern,
138            invoke: Arc::new(invoke),
139        }
140    }
141
142    /// Invokes the typed handler after the transport has decoded its envelope.
143    pub fn invoke<'a>(
144        &self,
145        container: &'a Container,
146        context: MessageContext,
147        payload: Value,
148    ) -> BoxFuture<'a, Result<Option<Value>>> {
149        (self.invoke)(container, context, payload)
150    }
151}
152
153/// Metadata for one dependency-injected microservice class.
154pub struct MicroserviceDef {
155    /// The provider definition used for normal module lifecycle management.
156    pub provider: ProviderDef,
157    handlers_fn: fn() -> Vec<MessageHandlerDef>,
158}
159
160impl MicroserviceDef {
161    /// Creates metadata for `T` and its generated handler factory.
162    pub fn of<T: Injectable>(handlers_fn: fn() -> Vec<MessageHandlerDef>) -> Self {
163        Self {
164            provider: ProviderDef::of::<T>(),
165            handlers_fn,
166        }
167    }
168
169    /// Materializes handler definitions for a runtime transport.
170    pub fn handlers(&self) -> Vec<MessageHandlerDef> {
171        (self.handlers_fn)()
172    }
173}
174
175/// Implemented by `#[microservice]` classes.
176pub trait Microservice: Injectable {
177    /// Returns provider and handler metadata for module registration.
178    fn definition() -> MicroserviceDef;
179}
180
181/// Resolves a handler's owning service as an erased singleton.
182///
183/// This helper is intentionally small; generated handler code resolves the
184/// concrete service itself so type checking remains at the declaration site.
185#[doc(hidden)]
186pub fn _assert_microservice_send_sync<T: Send + Sync + 'static>() {
187    let _ = std::marker::PhantomData::<Arc<T>>;
188}
caelix_core/microservice.rs

caelix_core/
microservice.rs
