a3s_boot/transport/
pattern.rs

1use super::{
2    ExecutionTransportGuard, ExecutionTransportInterceptor, IntoTransportReply, MessagePatternKind,
3    TransportContext, TransportGuard, TransportInterceptor, TransportMessage, TransportPipe,
4    TransportReply,
5};
6use crate::pipeline::{PipelineComponent, PipelineOverrides, ProviderEnhancerComponents};
7use crate::{
8    catch_errors, validate_json_value_with_options, BootError, BootErrorKind, BootRequest,
9    BoxFuture, CallHandler, ContextId, ContextIdFactory, ExecutionInterceptor, Guard, HttpMethod,
10    ModuleRef, Result, TransportExceptionFilter, Validate, ValidationOptions, ValidationSchema,
11};
12use serde::de::DeserializeOwned;
13use serde::Serialize;
14use serde_json::Value;
15use std::collections::BTreeMap;
16use std::future::Future;
17use std::sync::{Arc, Mutex};
18
19type TransportHandlerFuture = BoxFuture<'static, Result<Option<TransportReply>>>;
20type MessageValidator =
21    Arc<dyn Fn(TransportMessage, ValidationOptions) -> Result<TransportMessage> + Send + Sync>;
22
23trait TransportMessageHandler: Send + Sync + 'static {
24    fn call(&self, message: TransportMessage) -> TransportHandlerFuture;
25}
26
27struct TransportHandlerAdapter<H> {
28    handler: H,
29}
30
31impl<H, Fut, R> TransportMessageHandler for TransportHandlerAdapter<H>
32where
33    H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
34    Fut: Future<Output = Result<R>> + Send + 'static,
35    R: IntoTransportReply + Send + 'static,
36{
37    fn call(&self, message: TransportMessage) -> TransportHandlerFuture {
38        let future = (self.handler)(message);
39        Box::pin(async move { Ok(future.await?.into_transport_reply()) })
40    }
41}
42
43type ScopedTransportHandlerFactory =
44    dyn Fn(&ModuleRef) -> Result<Arc<dyn TransportMessageHandler>> + Send + Sync;
45
46#[derive(Clone)]
47enum TransportHandlerDefinition {
48    Static(Arc<dyn TransportMessageHandler>),
49    Scoped(Arc<ScopedTransportHandlerFactory>),
50}
51
52impl TransportHandlerDefinition {
53    fn resolve(
54        &self,
55        module_ref: Option<&ModuleRef>,
56        pattern: &str,
57    ) -> Result<Arc<dyn TransportMessageHandler>> {
58        match self {
59            Self::Static(handler) => Ok(Arc::clone(handler)),
60            Self::Scoped(factory) => {
61                let module_ref = module_ref.ok_or_else(|| {
62                    BootError::Internal(format!(
63                        "scoped transport message pattern `{pattern}` requires a declaring or default module context"
64                    ))
65                })?;
66                factory(module_ref)
67            }
68        }
69    }
70
71    fn is_scoped(&self) -> bool {
72        matches!(self, Self::Scoped(_))
73    }
74}
75
76#[derive(Default)]
77struct DispatchHandlerCache {
78    handler: Mutex<Option<Arc<dyn TransportMessageHandler>>>,
79}
80
81impl DispatchHandlerCache {
82    fn resolve(
83        &self,
84        definition: &TransportHandlerDefinition,
85        module_ref: Option<&ModuleRef>,
86        pattern: &str,
87    ) -> Result<Arc<dyn TransportMessageHandler>> {
88        let mut cached = self
89            .handler
90            .lock()
91            .unwrap_or_else(std::sync::PoisonError::into_inner);
92        if let Some(handler) = cached.as_ref() {
93            return Ok(Arc::clone(handler));
94        }
95
96        let handler = definition.resolve(module_ref, pattern)?;
97        *cached = Some(Arc::clone(&handler));
98        Ok(handler)
99    }
100}
101
102#[derive(Clone)]
103struct MessageDispatchState {
104    context_id: ContextId,
105    module_ref: Option<ModuleRef>,
106    handler: Arc<DispatchHandlerCache>,
107}
108
109impl MessageDispatchState {
110    fn new(module_ref: Option<&ModuleRef>) -> Self {
111        let context_id = ContextIdFactory::create();
112        let module_ref = module_ref.map(|module_ref| module_ref.context_scope(&context_id));
113        Self {
114            context_id,
115            module_ref,
116            handler: Arc::new(DispatchHandlerCache::default()),
117        }
118    }
119
120    fn request(&self) -> BootRequest {
121        let request = BootRequest::new(HttpMethod::Post, "/__transport");
122        match &self.module_ref {
123            Some(module_ref) => request.with_module_ref(module_ref.clone()),
124            None => request,
125        }
126    }
127}
128
129/// Framework-neutral message pattern handler definition.
130#[derive(Clone)]
131pub struct MessagePatternDefinition {
132    pattern: String,
133    kind: MessagePatternKind,
134    handler: TransportHandlerDefinition,
135    pipes: Vec<PipelineComponent<dyn TransportPipe>>,
136    guards: Vec<PipelineComponent<dyn TransportGuard>>,
137    interceptors: Vec<PipelineComponent<dyn TransportInterceptor>>,
138    filters: Vec<PipelineComponent<dyn TransportExceptionFilter>>,
139    validators: Vec<MessageValidator>,
140    validation_enabled: bool,
141    validation_disabled: bool,
142    validation_options: ValidationOptions,
143    metadata: BTreeMap<String, Value>,
144    module_name: Option<String>,
145    module_ref: Option<ModuleRef>,
146}
147
148impl MessagePatternDefinition {
149    pub fn request<H, Fut, R>(pattern: impl Into<String>, handler: H) -> Result<Self>
150    where
151        H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
152        Fut: Future<Output = Result<R>> + Send + 'static,
153        R: IntoTransportReply + Send + 'static,
154    {
155        Self::new(pattern, MessagePatternKind::RequestResponse, handler)
156    }
157
158    /// Build a request-response pattern whose handler is created from the
159    /// current message dispatch's dependency-injection scope.
160    pub fn request_scoped<F, H, Fut, R>(pattern: impl Into<String>, factory: F) -> Result<Self>
161    where
162        F: Fn(&ModuleRef) -> Result<H> + Send + Sync + 'static,
163        H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
164        Fut: Future<Output = Result<R>> + Send + 'static,
165        R: IntoTransportReply + Send + 'static,
166    {
167        Self::new_scoped(pattern, MessagePatternKind::RequestResponse, factory)
168    }
169
170    pub fn event<H, Fut>(pattern: impl Into<String>, handler: H) -> Result<Self>
171    where
172        H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
173        Fut: Future<Output = Result<()>> + Send + 'static,
174    {
175        Self::new(pattern, MessagePatternKind::Event, handler)
176    }
177
178    /// Build an event pattern whose handler is created from the current
179    /// message dispatch's dependency-injection scope.
180    pub fn event_scoped<F, H, Fut>(pattern: impl Into<String>, factory: F) -> Result<Self>
181    where
182        F: Fn(&ModuleRef) -> Result<H> + Send + Sync + 'static,
183        H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
184        Fut: Future<Output = Result<()>> + Send + 'static,
185    {
186        Self::new_scoped(pattern, MessagePatternKind::Event, factory)
187    }
188
189    pub fn request_json<T, H, Fut, R>(pattern: impl Into<String>, handler: H) -> Result<Self>
190    where
191        T: DeserializeOwned + Send + 'static,
192        H: Fn(T) -> Fut + Send + Sync + 'static,
193        Fut: Future<Output = Result<R>> + Send + 'static,
194        R: Serialize + Send + 'static,
195    {
196        Self::request(pattern, move |message: TransportMessage| {
197            let payload = message.data_as::<T>();
198            let future = payload.map(&handler);
199            async move {
200                let response = future?.await?;
201                TransportReply::json(&response)
202            }
203        })
204    }
205
206    pub fn request_validated_json<T, H, Fut, R>(
207        pattern: impl Into<String>,
208        handler: H,
209    ) -> Result<Self>
210    where
211        T: DeserializeOwned + Validate + Send + 'static,
212        H: Fn(T) -> Fut + Send + Sync + 'static,
213        Fut: Future<Output = Result<R>> + Send + 'static,
214        R: Serialize + Send + 'static,
215    {
216        Self::request(pattern, move |message: TransportMessage| {
217            let payload = message.validated_data::<T>();
218            let future = payload.map(&handler);
219            async move {
220                let response = future?.await?;
221                TransportReply::json(&response)
222            }
223        })
224    }
225
226    pub fn event_json<T, H, Fut>(pattern: impl Into<String>, handler: H) -> Result<Self>
227    where
228        T: DeserializeOwned + Send + 'static,
229        H: Fn(T) -> Fut + Send + Sync + 'static,
230        Fut: Future<Output = Result<()>> + Send + 'static,
231    {
232        Self::event(pattern, move |message: TransportMessage| {
233            let payload = message.data_as::<T>();
234            let future = payload.map(&handler);
235            async move { future?.await }
236        })
237    }
238
239    pub fn event_validated_json<T, H, Fut>(pattern: impl Into<String>, handler: H) -> Result<Self>
240    where
241        T: DeserializeOwned + Validate + Send + 'static,
242        H: Fn(T) -> Fut + Send + Sync + 'static,
243        Fut: Future<Output = Result<()>> + Send + 'static,
244    {
245        Self::event(pattern, move |message: TransportMessage| {
246            let payload = message.validated_data::<T>();
247            let future = payload.map(&handler);
248            async move { future?.await }
249        })
250    }
251
252    fn new<H, Fut, R>(
253        pattern: impl Into<String>,
254        kind: MessagePatternKind,
255        handler: H,
256    ) -> Result<Self>
257    where
258        H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
259        Fut: Future<Output = Result<R>> + Send + 'static,
260        R: IntoTransportReply + Send + 'static,
261    {
262        Self::from_handler(
263            pattern,
264            kind,
265            TransportHandlerDefinition::Static(Arc::new(TransportHandlerAdapter { handler })),
266        )
267    }
268
269    fn new_scoped<F, H, Fut, R>(
270        pattern: impl Into<String>,
271        kind: MessagePatternKind,
272        factory: F,
273    ) -> Result<Self>
274    where
275        F: Fn(&ModuleRef) -> Result<H> + Send + Sync + 'static,
276        H: Fn(TransportMessage) -> Fut + Send + Sync + 'static,
277        Fut: Future<Output = Result<R>> + Send + 'static,
278        R: IntoTransportReply + Send + 'static,
279    {
280        let factory = move |module_ref: &ModuleRef| {
281            let handler = factory(module_ref)?;
282            Ok(Arc::new(TransportHandlerAdapter { handler }) as Arc<dyn TransportMessageHandler>)
283        };
284        Self::from_handler(
285            pattern,
286            kind,
287            TransportHandlerDefinition::Scoped(Arc::new(factory)),
288        )
289    }
290
291    fn from_handler(
292        pattern: impl Into<String>,
293        kind: MessagePatternKind,
294        handler: TransportHandlerDefinition,
295    ) -> Result<Self> {
296        let pattern = pattern.into();
297        validate_pattern(&pattern)?;
298        Ok(Self {
299            pattern,
300            kind,
301            handler,
302            pipes: Vec::new(),
303            guards: Vec::new(),
304            interceptors: Vec::new(),
305            filters: Vec::new(),
306            validators: Vec::new(),
307            validation_enabled: false,
308            validation_disabled: false,
309            validation_options: ValidationOptions::default(),
310            metadata: BTreeMap::new(),
311            module_name: None,
312            module_ref: None,
313        })
314    }
315
316    pub fn pattern(&self) -> &str {
317        &self.pattern
318    }
319
320    pub fn kind(&self) -> MessagePatternKind {
321        self.kind
322    }
323
324    /// Return whether this pattern constructs its handler from each dispatch scope.
325    pub fn is_scoped(&self) -> bool {
326        self.handler.is_scoped()
327    }
328
329    pub fn module_name(&self) -> Option<&str> {
330        self.module_name.as_deref()
331    }
332
333    pub fn metadata(&self) -> &BTreeMap<String, Value> {
334        &self.metadata
335    }
336
337    pub fn metadata_value(&self, key: &str) -> Option<&Value> {
338        self.metadata.get(key)
339    }
340
341    pub fn with_metadata<V>(self, key: impl Into<String>, value: V) -> Result<Self>
342    where
343        V: Serialize,
344    {
345        let key = key.into();
346        let value = serde_json::to_value(value).map_err(|error| {
347            BootError::Internal(format!(
348                "failed to serialize message pattern metadata `{key}`: {error}"
349            ))
350        })?;
351        Ok(self.with_metadata_value(key, value))
352    }
353
354    pub fn with_metadata_value(mut self, key: impl Into<String>, value: Value) -> Self {
355        self.metadata.insert(key.into(), value);
356        self
357    }
358
359    pub fn with_pipe<P>(mut self, pipe: P) -> Self
360    where
361        P: TransportPipe,
362    {
363        self.pipes
364            .push(PipelineComponent::<dyn TransportPipe>::new(pipe));
365        self
366    }
367
368    pub fn with_guard<G>(mut self, guard: G) -> Self
369    where
370        G: TransportGuard,
371    {
372        self.guards
373            .push(PipelineComponent::<dyn TransportGuard>::new(guard));
374        self
375    }
376
377    pub fn with_execution_guard<G>(mut self, guard: G) -> Self
378    where
379        G: Guard,
380    {
381        self.guards
382            .push(PipelineComponent::<dyn TransportGuard>::new(
383                ExecutionTransportGuard { inner: guard },
384            ));
385        self
386    }
387
388    pub(crate) fn with_execution_pipeline_prefix(
389        mut self,
390        guards: &[Arc<dyn Guard>],
391        interceptors: &[Arc<dyn ExecutionInterceptor>],
392    ) -> Self {
393        self.guards = prepend_execution_guards(guards, self.guards);
394        self.interceptors = prepend_execution_interceptors(interceptors, self.interceptors);
395        self
396    }
397
398    pub(crate) fn with_guard_prefix(mut self, guards: &[Arc<dyn TransportGuard>]) -> Self {
399        let mut merged = guards
400            .iter()
401            .cloned()
402            .map(PipelineComponent::<dyn TransportGuard>::from_arc)
403            .collect::<Vec<_>>();
404        merged.extend(self.guards);
405        self.guards = merged;
406        self
407    }
408
409    pub(crate) fn with_interceptor_prefix(
410        mut self,
411        interceptors: &[Arc<dyn TransportInterceptor>],
412    ) -> Self {
413        let mut merged = interceptors
414            .iter()
415            .cloned()
416            .map(PipelineComponent::<dyn TransportInterceptor>::from_arc)
417            .collect::<Vec<_>>();
418        merged.extend(self.interceptors);
419        self.interceptors = merged;
420        self
421    }
422
423    pub(crate) fn with_pipe_prefix(mut self, pipes: &[Arc<dyn TransportPipe>]) -> Self {
424        let mut merged = pipes
425            .iter()
426            .cloned()
427            .map(PipelineComponent::<dyn TransportPipe>::from_arc)
428            .collect::<Vec<_>>();
429        merged.extend(self.pipes);
430        self.pipes = merged;
431        self
432    }
433
434    pub(crate) fn with_filter_prefix(
435        mut self,
436        filters: &[Arc<dyn TransportExceptionFilter>],
437    ) -> Self {
438        let mut merged = filters
439            .iter()
440            .cloned()
441            .map(PipelineComponent::<dyn TransportExceptionFilter>::from_arc)
442            .collect::<Vec<_>>();
443        merged.extend(self.filters);
444        self.filters = merged;
445        self
446    }
447
448    pub fn with_interceptor<I>(mut self, interceptor: I) -> Self
449    where
450        I: TransportInterceptor,
451    {
452        self.interceptors
453            .push(PipelineComponent::<dyn TransportInterceptor>::new(
454                interceptor,
455            ));
456        self
457    }
458
459    pub fn with_execution_interceptor<I>(mut self, interceptor: I) -> Self
460    where
461        I: ExecutionInterceptor,
462    {
463        self.interceptors
464            .push(PipelineComponent::<dyn TransportInterceptor>::new(
465                ExecutionTransportInterceptor { inner: interceptor },
466            ));
467        self
468    }
469
470    pub fn with_filter<F>(mut self, filter: F) -> Self
471    where
472        F: TransportExceptionFilter,
473    {
474        self.filters
475            .push(PipelineComponent::<dyn TransportExceptionFilter>::new(
476                filter,
477            ));
478        self
479    }
480
481    pub fn with_catch_filter<I, F>(self, kinds: I, filter: F) -> Self
482    where
483        I: IntoIterator<Item = BootErrorKind>,
484        F: TransportExceptionFilter,
485    {
486        self.with_filter(catch_errors(kinds, filter))
487    }
488
489    pub(crate) fn with_pipeline_overrides(mut self, overrides: &PipelineOverrides) -> Self {
490        overrides.apply_to_transport_pipes(&mut self.pipes);
491        overrides.apply_to_transport_guards(&mut self.guards);
492        overrides.apply_to_transport_interceptors(&mut self.interceptors);
493        overrides.apply_to_transport_filters(&mut self.filters);
494        self
495    }
496
497    pub(crate) fn with_provider_enhancer_prefix(
498        mut self,
499        enhancers: &ProviderEnhancerComponents,
500    ) -> Self {
501        let mut pipes = enhancers.transport_pipes.clone();
502        pipes.extend(self.pipes);
503        self.pipes = pipes;
504
505        let mut guards = enhancers.transport_guards.clone();
506        guards.extend(self.guards);
507        self.guards = guards;
508
509        let mut interceptors = enhancers.transport_interceptors.clone();
510        interceptors.extend(self.interceptors);
511        self.interceptors = interceptors;
512
513        let mut filters = enhancers.transport_filters.clone();
514        filters.extend(self.filters);
515        self.filters = filters;
516        self
517    }
518
519    pub fn with_validation(mut self) -> Self {
520        self.validation_enabled = true;
521        self.validation_disabled = false;
522        self
523    }
524
525    pub fn with_validation_options(mut self, options: ValidationOptions) -> Self {
526        self.validation_enabled = true;
527        self.validation_disabled = false;
528        self.validation_options = self.validation_options.merge(options);
529        self
530    }
531
532    pub fn without_validation(mut self) -> Self {
533        self.validation_enabled = false;
534        self.validation_disabled = true;
535        self
536    }
537
538    pub(crate) fn with_validation_prefix(
539        mut self,
540        validation_enabled: bool,
541        validation_options: ValidationOptions,
542    ) -> Self {
543        if !self.validation_disabled {
544            self.validation_enabled = validation_enabled || self.validation_enabled;
545            self.validation_options = validation_options.merge(self.validation_options);
546        }
547        self
548    }
549
550    pub fn with_payload_validation<T>(mut self) -> Self
551    where
552        T: DeserializeOwned + Validate + 'static,
553    {
554        self.validators.push(Arc::new(|message, _| {
555            message.validated_data::<T>().map(|_| message)
556        }));
557        self.with_validation()
558    }
559
560    pub fn with_payload_validation_options<T>(mut self, options: ValidationOptions) -> Self
561    where
562        T: DeserializeOwned + Serialize + Validate + ValidationSchema + 'static,
563    {
564        self.validators
565            .push(Arc::new(move |mut message, inherited_options| {
566                let options = inherited_options.merge(options);
567                let data = validate_json_value_with_options::<T>(
568                    message.data.clone(),
569                    options,
570                    "message property",
571                )?;
572                if options.transform || options.whitelist {
573                    message.data = data;
574                }
575                Ok(message)
576            }));
577        self.with_validation()
578    }
579
580    pub async fn dispatch(&self, message: TransportMessage) -> Result<Option<TransportReply>> {
581        let state = MessageDispatchState::new(self.module_ref.as_ref());
582        let context = TransportContext::new(self, &message.pattern, state.request());
583        match self
584            .dispatch_pipeline(message, context.clone(), state.clone())
585            .await
586        {
587            Ok(reply) => Ok(reply),
588            Err(error) => self.handle_error(context, error, &state.context_id).await,
589        }
590    }
591
592    async fn dispatch_pipeline(
593        &self,
594        message: TransportMessage,
595        context: TransportContext,
596        state: MessageDispatchState,
597    ) -> Result<Option<TransportReply>> {
598        if message.pattern != self.pattern {
599            return Err(BootError::NotFound(format!(
600                "message pattern {}",
601                message.pattern
602            )));
603        }
604
605        for guard in &self.guards {
606            let can_activate = guard
607                .resolve(&state.context_id)?
608                .can_activate(context.clone())
609                .await?;
610            if !can_activate {
611                return Err(BootError::Forbidden(format!(
612                    "message pattern {}",
613                    message.pattern
614                )));
615            }
616        }
617
618        let reply = self
619            .dispatch_interceptor_chain(0, context, message, state)
620            .await?;
621        Ok(if self.kind == MessagePatternKind::Event {
622            None
623        } else {
624            reply
625        })
626    }
627
628    fn dispatch_interceptor_chain<'a>(
629        &'a self,
630        index: usize,
631        context: TransportContext,
632        message: TransportMessage,
633        state: MessageDispatchState,
634    ) -> BoxFuture<'a, Result<Option<TransportReply>>> {
635        Box::pin(async move {
636            let Some(interceptor) = self.interceptors.get(index) else {
637                return self.dispatch_handler_pipeline(message, state).await;
638            };
639            let interceptor = interceptor.resolve(&state.context_id)?;
640
641            let next_context = context.clone();
642            let next_message = message.clone();
643            let next_state = state.clone();
644            let next = CallHandler::from_fn(move || {
645                self.dispatch_interceptor_chain(
646                    index + 1,
647                    next_context.clone(),
648                    next_message.clone(),
649                    next_state.clone(),
650                )
651            });
652            interceptor.intercept(context, next).await
653        })
654    }
655
656    async fn dispatch_handler_pipeline(
657        &self,
658        mut message: TransportMessage,
659        state: MessageDispatchState,
660    ) -> Result<Option<TransportReply>> {
661        for pipe in &self.pipes {
662            message = pipe.resolve(&state.context_id)?.transform(message).await?;
663        }
664
665        if self.validation_enabled {
666            for validator in &self.validators {
667                message = validator(message, self.validation_options)?;
668            }
669        }
670
671        let handler = state.handler.resolve(
672            &self.handler,
673            state.module_ref.as_ref(),
674            self.pattern.as_str(),
675        )?;
676        let mut reply = handler.call(message).await?;
677        if self.kind == MessagePatternKind::Event {
678            reply = None;
679        }
680        Ok(reply)
681    }
682
683    async fn handle_error(
684        &self,
685        context: TransportContext,
686        error: BootError,
687        context_id: &ContextId,
688    ) -> Result<Option<TransportReply>> {
689        for filter in self.filters.iter().rev() {
690            let filter = filter.resolve(context_id)?;
691            if let Some(response) = filter
692                .catch(context.clone(), error.clone_for_filter())
693                .await?
694            {
695                return Ok(if self.kind == MessagePatternKind::Event {
696                    None
697                } else {
698                    response.into_reply()
699                });
700            }
701        }
702        Err(error)
703    }
704
705    pub(crate) fn with_module_name(mut self, module_name: &str) -> Self {
706        self.module_name = Some(module_name.to_string());
707        self
708    }
709
710    pub(crate) fn with_module_ref(mut self, module_ref: ModuleRef) -> Self {
711        self.module_ref = Some(module_ref);
712        self
713    }
714
715    pub(crate) fn with_default_module_ref(mut self, module_ref: ModuleRef) -> Self {
716        if self.module_ref.is_none() {
717            self.module_ref = Some(module_ref);
718        }
719        self
720    }
721}
722
723fn prepend_execution_guards(
724    prefix: &[Arc<dyn Guard>],
725    values: Vec<PipelineComponent<dyn TransportGuard>>,
726) -> Vec<PipelineComponent<dyn TransportGuard>> {
727    let mut merged = prefix
728        .iter()
729        .cloned()
730        .map(|guard| {
731            PipelineComponent::<dyn TransportGuard>::new(ExecutionTransportGuard { inner: guard })
732        })
733        .collect::<Vec<_>>();
734    merged.extend(values);
735    merged
736}
737
738fn prepend_execution_interceptors(
739    prefix: &[Arc<dyn ExecutionInterceptor>],
740    values: Vec<PipelineComponent<dyn TransportInterceptor>>,
741) -> Vec<PipelineComponent<dyn TransportInterceptor>> {
742    let mut merged = prefix
743        .iter()
744        .cloned()
745        .map(|interceptor| {
746            PipelineComponent::<dyn TransportInterceptor>::new(ExecutionTransportInterceptor {
747                inner: interceptor,
748            })
749        })
750        .collect::<Vec<_>>();
751    merged.extend(values);
752    merged
753}
754
755fn validate_pattern(pattern: &str) -> Result<()> {
756    if pattern.trim().is_empty() {
757        return Err(BootError::BadRequest(
758            "message pattern cannot be empty".to_string(),
759        ));
760    }
761    Ok(())
762}
a3s_boot/transport/pattern.rs

a3s_boot/transport/
pattern.rs
