a3s_boot/transport/
mod.rs

1use crate::{
2    validate_value, BootError, BootRequest, BoxFuture, CallHandler, ExecutionContext,
3    ExecutionInterceptor, ExecutionTransportKind, Guard, Result, Validate,
4};
5use serde::de::DeserializeOwned;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use std::future::Future;
9use std::ops::Deref;
10
11mod pattern;
12
13pub use self::pattern::MessagePatternDefinition;
14
15#[cfg(feature = "grpc-transport")]
16mod grpc;
17#[cfg(feature = "kafka-transport")]
18mod kafka;
19#[cfg(feature = "mqtt-transport")]
20mod mqtt;
21#[cfg(feature = "nats-transport")]
22mod nats;
23#[cfg(feature = "rabbitmq-transport")]
24mod rabbitmq;
25#[cfg(feature = "redis-transport")]
26mod redis;
27#[cfg(feature = "tcp-transport")]
28mod tcp;
29
30#[cfg(feature = "grpc-transport")]
31pub use self::grpc::{GrpcTransport, GrpcTransportClient, GrpcTransportOptions};
32#[cfg(feature = "kafka-transport")]
33pub use self::kafka::{KafkaTransport, KafkaTransportClient, KafkaTransportOptions};
34#[cfg(feature = "mqtt-transport")]
35pub use self::mqtt::{MqttTransport, MqttTransportClient, MqttTransportOptions, MqttTransportQoS};
36#[cfg(feature = "nats-transport")]
37pub use self::nats::{NatsTransport, NatsTransportClient, NatsTransportOptions};
38#[cfg(feature = "rabbitmq-transport")]
39pub use self::rabbitmq::{RabbitMqTransport, RabbitMqTransportClient, RabbitMqTransportOptions};
40#[cfg(feature = "redis-transport")]
41pub use self::redis::{RedisTransport, RedisTransportClient, RedisTransportOptions};
42#[cfg(any(
43    feature = "grpc-transport",
44    feature = "kafka-transport",
45    feature = "mqtt-transport",
46    feature = "nats-transport",
47    feature = "rabbitmq-transport",
48    feature = "redis-transport",
49    feature = "tcp-transport"
50))]
51pub(super) fn transport_error_from_status(status: u16, message: String) -> BootError {
52    BootError::from_http_status(status, message)
53}
54
55#[cfg(feature = "tcp-transport")]
56pub use tcp::{TcpTransport, TcpTransportClient, TcpTransportOptions};
57
58/// Adapter-neutral microservice transport message.
59#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
60pub struct TransportMessage {
61    pub pattern: String,
62    #[serde(default)]
63    pub data: Value,
64}
65
66impl TransportMessage {
67    pub fn new(pattern: impl Into<String>, data: impl Into<Value>) -> Self {
68        Self {
69            pattern: pattern.into(),
70            data: data.into(),
71        }
72    }
73
74    pub fn pattern(&self) -> &str {
75        &self.pattern
76    }
77
78    pub fn data(&self) -> &Value {
79        &self.data
80    }
81
82    pub fn text(pattern: impl Into<String>, data: impl Into<String>) -> Self {
83        Self::new(pattern, Value::String(data.into()))
84    }
85
86    pub fn json<T>(pattern: impl Into<String>, data: &T) -> Result<Self>
87    where
88        T: Serialize,
89    {
90        Ok(Self::new(
91            pattern,
92            serde_json::to_value(data).map_err(|err| BootError::Internal(err.to_string()))?,
93        ))
94    }
95
96    pub fn data_as<T>(&self) -> Result<T>
97    where
98        T: DeserializeOwned,
99    {
100        serde_json::from_value(self.data.clone())
101            .map_err(|err| BootError::BadRequest(err.to_string()))
102    }
103
104    pub fn data_field(&self, name: &str) -> Result<Option<Value>> {
105        let Value::Object(fields) = &self.data else {
106            return Err(BootError::BadRequest(
107                "expected JSON object transport data".to_string(),
108            ));
109        };
110
111        Ok(fields.get(name).filter(|value| !value.is_null()).cloned())
112    }
113
114    pub fn data_field_as<T>(&self, name: &str) -> Result<T>
115    where
116        T: DeserializeOwned,
117    {
118        let Some(value) = self.data_field(name)? else {
119            return Err(BootError::BadRequest(format!(
120                "missing transport data field: {name}"
121            )));
122        };
123        deserialize_data_field("transport data field", name, value)
124    }
125
126    pub fn optional_data_field_as<T>(&self, name: &str) -> Result<Option<T>>
127    where
128        T: DeserializeOwned,
129    {
130        self.data_field(name)?
131            .map(|value| deserialize_data_field("transport data field", name, value))
132            .transpose()
133    }
134
135    pub fn data_field_string(&self, name: &str) -> Result<String> {
136        let Some(value) = self.data_field(name)? else {
137            return Err(BootError::BadRequest(format!(
138                "missing transport data field: {name}"
139            )));
140        };
141        data_field_value_to_string(value)
142    }
143
144    pub fn optional_data_field_string(&self, name: &str) -> Result<Option<String>> {
145        self.data_field(name)?
146            .map(data_field_value_to_string)
147            .transpose()
148    }
149
150    pub fn validated_data<T>(&self) -> Result<T>
151    where
152        T: DeserializeOwned + Validate,
153    {
154        validate_value(self.data_as::<T>()?)
155    }
156}
157
158fn deserialize_data_field<T>(label: &str, name: &str, value: Value) -> Result<T>
159where
160    T: DeserializeOwned,
161{
162    serde_json::from_value(value)
163        .map_err(|error| BootError::BadRequest(format!("invalid {label} {name}: {error}")))
164}
165
166fn data_field_value_to_string(value: Value) -> Result<String> {
167    match value {
168        Value::String(value) => Ok(value),
169        Value::Bool(value) => Ok(value.to_string()),
170        Value::Number(value) => Ok(value.to_string()),
171        Value::Array(_) | Value::Object(_) => {
172            serde_json::to_string(&value).map_err(|error| BootError::BadRequest(error.to_string()))
173        }
174        Value::Null => Ok("null".to_string()),
175    }
176}
177
178/// Reply returned by request-response message pattern handlers.
179#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
180pub struct TransportReply {
181    #[serde(default)]
182    pub data: Value,
183}
184
185impl TransportReply {
186    pub fn new(data: impl Into<Value>) -> Self {
187        Self { data: data.into() }
188    }
189
190    pub fn data(&self) -> &Value {
191        &self.data
192    }
193
194    pub fn text(data: impl Into<String>) -> Self {
195        Self::new(Value::String(data.into()))
196    }
197
198    pub fn json<T>(data: &T) -> Result<Self>
199    where
200        T: Serialize,
201    {
202        Ok(Self::new(
203            serde_json::to_value(data).map_err(|err| BootError::Internal(err.to_string()))?,
204        ))
205    }
206
207    pub fn data_as<T>(&self) -> Result<T>
208    where
209        T: DeserializeOwned,
210    {
211        serde_json::from_value(self.data.clone())
212            .map_err(|err| BootError::BadRequest(err.to_string()))
213    }
214}
215
216/// Return value accepted by request-response message pattern handlers.
217pub trait IntoTransportReply {
218    fn into_transport_reply(self) -> Option<TransportReply>;
219}
220
221impl IntoTransportReply for TransportReply {
222    fn into_transport_reply(self) -> Option<TransportReply> {
223        Some(self)
224    }
225}
226
227impl IntoTransportReply for Option<TransportReply> {
228    fn into_transport_reply(self) -> Option<TransportReply> {
229        self
230    }
231}
232
233impl IntoTransportReply for Value {
234    fn into_transport_reply(self) -> Option<TransportReply> {
235        Some(TransportReply::new(self))
236    }
237}
238
239impl IntoTransportReply for Option<Value> {
240    fn into_transport_reply(self) -> Option<TransportReply> {
241        self.map(TransportReply::new)
242    }
243}
244
245impl IntoTransportReply for () {
246    fn into_transport_reply(self) -> Option<TransportReply> {
247        None
248    }
249}
250
251#[derive(Debug, Clone, Copy, PartialEq, Eq)]
252pub enum MessagePatternKind {
253    RequestResponse,
254    Event,
255}
256
257impl MessagePatternKind {
258    pub fn as_str(self) -> &'static str {
259        match self {
260            Self::RequestResponse => "request-response",
261            Self::Event => "event",
262        }
263    }
264}
265
266/// Context available to transport guards and interceptors.
267#[derive(Debug, Clone)]
268pub struct TransportContext {
269    pub pattern: String,
270    pub kind: MessagePatternKind,
271    pub module_name: Option<String>,
272    execution_context: ExecutionContext,
273}
274
275impl TransportContext {
276    fn new(definition: &MessagePatternDefinition, pattern: &str, request: BootRequest) -> Self {
277        let pattern = pattern.to_string();
278        let kind = definition.kind();
279        let module_name = definition.module_name().map(str::to_string);
280        let metadata = definition.metadata().clone();
281        let execution_context = ExecutionContext::transport(
282            request,
283            pattern.clone(),
284            ExecutionTransportKind::from(kind),
285            module_name.clone(),
286            metadata,
287        );
288        Self {
289            pattern,
290            kind,
291            module_name,
292            execution_context,
293        }
294    }
295
296    pub fn execution_context(&self) -> &ExecutionContext {
297        &self.execution_context
298    }
299
300    pub fn into_execution_context(self) -> ExecutionContext {
301        self.execution_context
302    }
303}
304
305impl Deref for TransportContext {
306    type Target = ExecutionContext;
307
308    fn deref(&self) -> &Self::Target {
309        self.execution_context()
310    }
311}
312
313impl From<MessagePatternKind> for ExecutionTransportKind {
314    fn from(kind: MessagePatternKind) -> Self {
315        match kind {
316            MessagePatternKind::RequestResponse => Self::RequestResponse,
317            MessagePatternKind::Event => Self::Event,
318        }
319    }
320}
321
322/// Message transformation hook for transport message patterns.
323pub trait TransportPipe: Send + Sync + 'static {
324    fn transform(&self, message: TransportMessage) -> BoxFuture<'static, Result<TransportMessage>>;
325}
326
327impl<F, Fut> TransportPipe for F
328where
329    F: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
330    Fut: Future<Output = Result<TransportMessage>> + Send + 'static,
331{
332    fn transform(&self, message: TransportMessage) -> BoxFuture<'static, Result<TransportMessage>> {
333        Box::pin(self(message))
334    }
335}
336
337/// Authorization hook for transport message patterns.
338pub trait TransportGuard: Send + Sync + 'static {
339    fn can_activate(&self, context: TransportContext) -> BoxFuture<'static, Result<bool>>;
340}
341
342impl<F, Fut> TransportGuard for F
343where
344    F: Fn(TransportContext) -> Fut + Send + Sync + 'static,
345    Fut: Future<Output = Result<bool>> + Send + 'static,
346{
347    fn can_activate(&self, context: TransportContext) -> BoxFuture<'static, Result<bool>> {
348        Box::pin(self(context))
349    }
350}
351
352struct ExecutionTransportGuard<G> {
353    inner: G,
354}
355
356impl<G> TransportGuard for ExecutionTransportGuard<G>
357where
358    G: Guard,
359{
360    fn can_activate(&self, context: TransportContext) -> BoxFuture<'static, Result<bool>> {
361        self.inner.can_activate(context.into_execution_context())
362    }
363}
364
365/// Around-handler hook for transport message patterns.
366pub trait TransportInterceptor: Send + Sync + 'static {
367    /// Run around the remaining transport pipeline.
368    ///
369    /// Override this method to recover downstream errors, retry the remaining
370    /// pipeline, or return a reply without calling `next`. The default
371    /// implementation preserves the legacy `before` and `after` hook behavior.
372    fn intercept<'a>(
373        &'a self,
374        context: TransportContext,
375        next: CallHandler<'a, Option<TransportReply>>,
376    ) -> BoxFuture<'a, Result<Option<TransportReply>>> {
377        Box::pin(async move {
378            self.before(context.clone()).await?;
379            let reply = next.handle().await?;
380            self.after(context, reply).await
381        })
382    }
383
384    fn before(&self, _context: TransportContext) -> BoxFuture<'static, Result<()>> {
385        Box::pin(async { Ok(()) })
386    }
387
388    fn after(
389        &self,
390        _context: TransportContext,
391        reply: Option<TransportReply>,
392    ) -> BoxFuture<'static, Result<Option<TransportReply>>> {
393        Box::pin(async move { Ok(reply) })
394    }
395}
396
397struct ExecutionTransportInterceptor<I> {
398    inner: I,
399}
400
401impl<I> TransportInterceptor for ExecutionTransportInterceptor<I>
402where
403    I: ExecutionInterceptor,
404{
405    fn before(&self, context: TransportContext) -> BoxFuture<'static, Result<()>> {
406        self.inner.before(context.into_execution_context())
407    }
408
409    fn after(
410        &self,
411        context: TransportContext,
412        reply: Option<TransportReply>,
413    ) -> BoxFuture<'static, Result<Option<TransportReply>>> {
414        let future = self.inner.after(context.into_execution_context());
415        Box::pin(async move {
416            future.await?;
417            Ok(reply)
418        })
419    }
420}
421
422/// Adapter trait for message transports such as in-process, Redis, NATS, or Kafka.
423pub trait MessageTransport {
424    type Output;
425
426    fn build(&self, app: crate::BootApplication) -> Result<Self::Output>;
427
428    fn serve(&self, app: crate::BootApplication) -> BoxFuture<'static, Result<()>>;
429}
430
431/// In-process transport useful for tests and single-process message dispatch.
432#[derive(Debug, Clone, Copy, Default)]
433pub struct InProcessTransport;
434
435impl InProcessTransport {
436    pub fn new() -> Self {
437        Self
438    }
439}
440
441impl MessageTransport for InProcessTransport {
442    type Output = InProcessTransportClient;
443
444    fn build(&self, app: crate::BootApplication) -> Result<Self::Output> {
445        Ok(InProcessTransportClient { app })
446    }
447
448    fn serve(&self, _app: crate::BootApplication) -> BoxFuture<'static, Result<()>> {
449        Box::pin(async { Ok(()) })
450    }
451}
452
453/// In-process message client backed by a resolved Boot application.
454#[derive(Clone)]
455pub struct InProcessTransportClient {
456    app: crate::BootApplication,
457}
458
459impl InProcessTransportClient {
460    pub async fn send(&self, message: TransportMessage) -> Result<Option<TransportReply>> {
461        self.app.dispatch_message(message).await
462    }
463
464    pub async fn emit(&self, message: TransportMessage) -> Result<()> {
465        self.app.emit_message(message).await
466    }
467}
a3s_boot/transport/mod.rs

a3s_boot/transport/
mod.rs
