nautilus_event_store/capture/
registry.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! The bus capture allow-list.
17//!
18//! [`EncoderRegistry`] is the allow-list the [`crate::capture::BusCaptureAdapter`] consults
19//! at every dispatch boundary. A message whose Rust type has no registered encoder is not
20//! captured: the SPEC names a closed list of state-affecting topics, and silent capture of
21//! out-of-allow-list types would produce entries the verifier process cannot decode.
22//!
23//! Registration binds three things:
24//!
25//! 1. The Rust type the encoder consumes (used as the [`std::any::TypeId`] lookup key).
26//! 2. The canonical [`crate::PayloadType`] tag stamped on every entry the encoder produces.
27//! 3. The encoder closure that produces the payload bytes plus sidecar [`crate::IndexKey`]s.
28//!
29//! A registration may additionally carry an identity extractor: production dispatch pushes
30//! the same typed message through multiple tap-visible boundaries (an order event is sent
31//! to the portfolio endpoint and published on its strategy topic), and the adapter uses the
32//! extracted identity to capture each logical message exactly once.
33
34use std::{
35    any::{Any, TypeId},
36    collections::HashMap,
37    fmt::Debug,
38    marker::PhantomData,
39    sync::Arc,
40};
41
42use nautilus_core::UUID4;
43
44use crate::{
45    capture::encoder::{Encode, EncodeError, EncodedPayload, TypedEncoder},
46    entry::PayloadType,
47    headers::Headers,
48};
49
50/// Extracts correlation [`Headers`] from a captured message at the dispatch boundary.
51///
52/// The encoder boundary deliberately omits headers ([`EncodedPayload`] documents this), so
53/// the bus tap consults the registry-registered extractor instead. Registration without an
54/// explicit extractor falls back to [`Headers::empty`]; that keeps capture working for
55/// allow-list types whose underlying struct has not yet grown `correlation_id` /
56/// `causation_id`
57/// fields (header propagation lands incrementally per the SPEC).
58pub trait HeadersExtractor: Send + Sync {
59    /// Returns the headers carried by `message`. Implementations downcast to the concrete
60    /// type they were registered for and read the relevant fields; mismatched types yield
61    /// [`Headers::empty`] so a stale registration cannot crash the tap.
62    fn extract(&self, message: &dyn Any) -> Headers;
63}
64
65/// Typed adapter that downcasts to `T` and forwards to a `Fn(&T) -> Headers` closure.
66pub struct TypedHeadersExtractor<T: 'static, F> {
67    func: F,
68    _phantom: PhantomData<fn(&T)>,
69}
70
71impl<T: 'static, F> TypedHeadersExtractor<T, F>
72where
73    F: Fn(&T) -> Headers + Send + Sync,
74{
75    /// Wraps `func` as a typed headers extractor for `T`.
76    #[must_use]
77    pub const fn new(func: F) -> Self {
78        Self {
79            func,
80            _phantom: PhantomData,
81        }
82    }
83}
84
85impl<T: 'static, F> Debug for TypedHeadersExtractor<T, F> {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct(stringify!(TypedHeadersExtractor))
88            .field("type", &std::any::type_name::<T>())
89            .finish_non_exhaustive()
90    }
91}
92
93impl<T: 'static, F> HeadersExtractor for TypedHeadersExtractor<T, F>
94where
95    F: Fn(&T) -> Headers + Send + Sync,
96{
97    fn extract(&self, message: &dyn Any) -> Headers {
98        message
99            .downcast_ref::<T>()
100            .map(&self.func)
101            .unwrap_or_default()
102    }
103}
104
105// Extractor that always returns `Headers::empty()`; the default for registrations that
106// have no explicit extractor.
107#[derive(Debug, Default)]
108struct EmptyHeadersExtractor;
109
110impl HeadersExtractor for EmptyHeadersExtractor {
111    fn extract(&self, _: &dyn Any) -> Headers {
112        Headers::empty()
113    }
114}
115
116/// Extracts the logical message identity (event id, command id) from a captured message.
117///
118/// The adapter uses the identity to capture a message exactly once when dispatch pushes
119/// it through multiple tap-visible boundaries. Types without an extractor are captured
120/// per dispatch, which is correct only for types that reach the tap on a single boundary.
121pub trait IdentityExtractor: Send + Sync {
122    /// Returns the identity carried by `message`. Mismatched types yield `None` so a
123    /// stale registration cannot crash the tap.
124    fn extract(&self, message: &dyn Any) -> Option<UUID4>;
125}
126
127/// Typed adapter that downcasts to `T` and forwards to a `Fn(&T) -> Option<UUID4>` closure.
128pub struct TypedIdentityExtractor<T: 'static, F> {
129    func: F,
130    _phantom: PhantomData<fn(&T)>,
131}
132
133impl<T: 'static, F> TypedIdentityExtractor<T, F>
134where
135    F: Fn(&T) -> Option<UUID4> + Send + Sync,
136{
137    /// Wraps `func` as a typed identity extractor for `T`.
138    #[must_use]
139    pub const fn new(func: F) -> Self {
140        Self {
141            func,
142            _phantom: PhantomData,
143        }
144    }
145}
146
147impl<T: 'static, F> Debug for TypedIdentityExtractor<T, F> {
148    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
149        f.debug_struct(stringify!(TypedIdentityExtractor))
150            .field("type", &std::any::type_name::<T>())
151            .finish_non_exhaustive()
152    }
153}
154
155impl<T: 'static, F> IdentityExtractor for TypedIdentityExtractor<T, F>
156where
157    F: Fn(&T) -> Option<UUID4> + Send + Sync,
158{
159    fn extract(&self, message: &dyn Any) -> Option<UUID4> {
160        message.downcast_ref::<T>().and_then(&self.func)
161    }
162}
163
164// One allow-list entry: canonical payload tag, encoder, header extractor, and optional
165// identity extractor.
166#[derive(Clone)]
167struct Registered {
168    payload_type: PayloadType,
169    encoder: Arc<dyn Encode>,
170    headers: Arc<dyn HeadersExtractor>,
171    identity: Option<Arc<dyn IdentityExtractor>>,
172}
173
174impl Debug for Registered {
175    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
176        f.debug_struct(stringify!(Registered))
177            .field("payload_type", &self.payload_type.as_str())
178            .finish_non_exhaustive()
179    }
180}
181
182/// Allow-list of capturable Rust message types and their encoders.
183///
184/// Registration is keyed by [`std::any::TypeId`]; the adapter dispatches by `TypeId` so
185/// callers can capture a typed message without naming the encoder concretely. A type
186/// registered twice replaces the prior entry: the SPEC's encoder rules require one
187/// canonical mapping per Rust type, and the registry would otherwise hide the conflict.
188#[derive(Clone, Debug, Default)]
189pub struct EncoderRegistry {
190    by_type: HashMap<TypeId, Registered>,
191}
192
193impl EncoderRegistry {
194    /// Creates a new empty registry.
195    #[must_use]
196    pub fn new() -> Self {
197        Self::default()
198    }
199
200    /// Registers `func` as the encoder for `T` and stamps every captured entry with
201    /// `payload_type`. Captures yield [`Headers::empty`] until [`Self::register_headers`]
202    /// records an extractor for `T`.
203    ///
204    /// Replaces any encoder previously registered for `T`; capture flows hold the
205    /// registry as `Arc<EncoderRegistry>` so registration must happen before the adapter
206    /// is constructed.
207    pub fn register<T, F>(&mut self, payload_type: PayloadType, func: F)
208    where
209        T: 'static,
210        F: Fn(&T) -> Result<EncodedPayload, EncodeError> + Send + Sync + 'static,
211    {
212        let encoder: Arc<dyn Encode> = Arc::new(TypedEncoder::<T, F>::new(func));
213        let headers = self
214            .preserved_headers::<T>()
215            .unwrap_or_else(|| Arc::new(EmptyHeadersExtractor) as Arc<dyn HeadersExtractor>);
216        let identity = self.preserved_identity::<T>();
217        self.by_type.insert(
218            TypeId::of::<T>(),
219            Registered {
220                payload_type,
221                encoder,
222                headers,
223                identity,
224            },
225        );
226    }
227
228    /// Registers `func` as the encoder for `T` and `headers_fn` as the matching headers
229    /// extractor in one call.
230    pub fn register_with_headers<T, F, H>(
231        &mut self,
232        payload_type: PayloadType,
233        func: F,
234        headers_fn: H,
235    ) where
236        T: 'static,
237        F: Fn(&T) -> Result<EncodedPayload, EncodeError> + Send + Sync + 'static,
238        H: Fn(&T) -> Headers + Send + Sync + 'static,
239    {
240        let encoder: Arc<dyn Encode> = Arc::new(TypedEncoder::<T, F>::new(func));
241        let headers: Arc<dyn HeadersExtractor> =
242            Arc::new(TypedHeadersExtractor::<T, H>::new(headers_fn));
243        let identity = self.preserved_identity::<T>();
244        self.by_type.insert(
245            TypeId::of::<T>(),
246            Registered {
247                payload_type,
248                encoder,
249                headers,
250                identity,
251            },
252        );
253    }
254
255    /// Registers an already-built [`Encode`] implementer for `T`. Captures yield
256    /// [`Headers::empty`] until [`Self::register_headers`] records an extractor.
257    ///
258    /// Useful when the encoder owns state (e.g., a schema cache) the closure form cannot
259    /// express ergonomically.
260    pub fn register_encoder<T: 'static>(
261        &mut self,
262        payload_type: PayloadType,
263        encoder: Arc<dyn Encode>,
264    ) {
265        let headers = self
266            .preserved_headers::<T>()
267            .unwrap_or_else(|| Arc::new(EmptyHeadersExtractor) as Arc<dyn HeadersExtractor>);
268        let identity = self.preserved_identity::<T>();
269        self.by_type.insert(
270            TypeId::of::<T>(),
271            Registered {
272                payload_type,
273                encoder,
274                headers,
275                identity,
276            },
277        );
278    }
279
280    /// Registers `headers_fn` as the headers extractor for `T`.
281    ///
282    /// Call after [`Self::register`] when the encoder is preregistered through a shared
283    /// helper but the call site wants a typed headers extractor. Replaces any prior
284    /// extractor for `T`. Returns silently when `T` has no encoder registered: callers
285    /// that care about the contract should rely on [`Self::contains`].
286    pub fn register_headers<T, H>(&mut self, headers_fn: H)
287    where
288        T: 'static,
289        H: Fn(&T) -> Headers + Send + Sync + 'static,
290    {
291        if let Some(reg) = self.by_type.get_mut(&TypeId::of::<T>()) {
292            reg.headers = Arc::new(TypedHeadersExtractor::<T, H>::new(headers_fn));
293        }
294    }
295
296    /// Registers `identity_fn` as the identity extractor for `T`.
297    ///
298    /// Call after [`Self::register`]. Replaces any prior extractor for `T`. Returns
299    /// silently when `T` has no encoder registered: callers that care about the contract
300    /// should rely on [`Self::contains`].
301    pub fn register_identity<T, F>(&mut self, identity_fn: F)
302    where
303        T: 'static,
304        F: Fn(&T) -> Option<UUID4> + Send + Sync + 'static,
305    {
306        if let Some(reg) = self.by_type.get_mut(&TypeId::of::<T>()) {
307            reg.identity = Some(Arc::new(TypedIdentityExtractor::<T, F>::new(identity_fn)));
308        }
309    }
310
311    fn preserved_headers<T: 'static>(&self) -> Option<Arc<dyn HeadersExtractor>> {
312        self.by_type
313            .get(&TypeId::of::<T>())
314            .map(|reg| Arc::clone(&reg.headers))
315    }
316
317    fn preserved_identity<T: 'static>(&self) -> Option<Arc<dyn IdentityExtractor>> {
318        self.by_type
319            .get(&TypeId::of::<T>())
320            .and_then(|reg| reg.identity.clone())
321    }
322
323    /// Returns the number of registered encoders.
324    #[must_use]
325    pub fn len(&self) -> usize {
326        self.by_type.len()
327    }
328
329    /// Returns whether the registry is empty.
330    #[must_use]
331    pub fn is_empty(&self) -> bool {
332        self.by_type.is_empty()
333    }
334
335    /// Returns whether an encoder is registered for `T`.
336    #[must_use]
337    pub fn contains<T: 'static>(&self) -> bool {
338        self.by_type.contains_key(&TypeId::of::<T>())
339    }
340
341    /// Encodes `message` if a typed encoder is registered for `T`.
342    ///
343    /// Returns `Ok(None)` when no encoder is registered for `T` so the adapter can drop
344    /// the message at the dispatch boundary without surfacing it as an error: the
345    /// allow-list is the source of truth for the captured surface, and out-of-list
346    /// messages are non-state-affecting by definition.
347    ///
348    /// # Errors
349    ///
350    /// Returns the encoder's [`EncodeError`] when an encoder is registered but rejects
351    /// the message.
352    pub fn encode<T: 'static>(
353        &self,
354        message: &T,
355    ) -> Result<Option<(PayloadType, EncodedPayload)>, EncodeError> {
356        let Some(reg) = self.by_type.get(&TypeId::of::<T>()) else {
357            return Ok(None);
358        };
359
360        let encoded = reg.encoder.encode(message as &dyn Any)?;
361        let payload_type = encoded.payload_type.unwrap_or(reg.payload_type);
362        Ok(Some((payload_type, encoded)))
363    }
364
365    /// Encodes a type-erased `message` if an encoder is registered for the concrete type.
366    ///
367    /// Mirror of [`Self::encode`] for `&dyn Any` callers; the bus tap reaches the
368    /// capture path through this entry point because dispatch returns a `&dyn Any` and
369    /// the static type is not in scope.
370    ///
371    /// # Errors
372    ///
373    /// Returns the encoder's [`EncodeError`] when an encoder is registered for the
374    /// concrete type but rejects the message.
375    pub fn encode_any(
376        &self,
377        message: &dyn Any,
378    ) -> Result<Option<(PayloadType, EncodedPayload)>, EncodeError> {
379        let Some(reg) = self.by_type.get(&message.type_id()) else {
380            return Ok(None);
381        };
382
383        let encoded = reg.encoder.encode(message)?;
384        let payload_type = encoded.payload_type.unwrap_or(reg.payload_type);
385        Ok(Some((payload_type, encoded)))
386    }
387
388    /// Returns the headers carried by `message` if a registration exists for its concrete
389    /// type. Returns `None` when no encoder is registered: the adapter then drops the
390    /// message without further work.
391    #[must_use]
392    pub fn headers_for_any(&self, message: &dyn Any) -> Option<Headers> {
393        self.by_type
394            .get(&message.type_id())
395            .map(|reg| reg.headers.extract(message))
396    }
397
398    /// Returns the logical identity carried by `message` if its registration has an
399    /// identity extractor. Returns `None` when the type is unregistered or carries no
400    /// extractor; the adapter then captures per dispatch.
401    #[must_use]
402    pub fn identity_for_any(&self, message: &dyn Any) -> Option<UUID4> {
403        self.by_type
404            .get(&message.type_id())
405            .and_then(|reg| reg.identity.as_ref())
406            .and_then(|identity| identity.extract(message))
407    }
408}
409
410#[cfg(test)]
411mod tests {
412    use std::sync::Arc;
413
414    use bytes::Bytes;
415    use rstest::rstest;
416    use ustr::Ustr;
417
418    use super::*;
419
420    #[derive(Debug)]
421    struct Sample(u8);
422
423    #[derive(Debug)]
424    struct Other;
425
426    #[derive(Debug)]
427    struct StatefulEncoder {
428        prefix: u8,
429    }
430
431    impl Encode for StatefulEncoder {
432        fn encode(&self, message: &dyn Any) -> Result<EncodedPayload, EncodeError> {
433            let sample = message
434                .downcast_ref::<Sample>()
435                .ok_or(EncodeError::TypeMismatch {
436                    expected: std::any::type_name::<Sample>(),
437                })?;
438            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
439                self.prefix,
440                sample.0,
441            ])))
442        }
443    }
444
445    #[rstest]
446    fn unknown_type_returns_none() {
447        let registry = EncoderRegistry::new();
448
449        assert!(registry.encode(&Sample(1)).expect("encode").is_none());
450        assert!(!registry.contains::<Sample>());
451    }
452
453    #[rstest]
454    fn registered_type_returns_payload_type_and_payload() {
455        let mut registry = EncoderRegistry::new();
456        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
457            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
458                s.0,
459            ])))
460        });
461
462        let (tag, encoded) = registry.encode(&Sample(9)).expect("encode").expect("hit");
463
464        assert_eq!(tag.as_str(), "Sample");
465        assert_eq!(encoded.payload.as_ref(), &[9]);
466        assert!(registry.contains::<Sample>());
467        assert_eq!(registry.len(), 1);
468    }
469
470    #[rstest]
471    fn re_registering_replaces_prior_encoder() {
472        let mut registry = EncoderRegistry::new();
473        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
474            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
475                s.0,
476            ])))
477        });
478        registry.register::<Sample, _>(Ustr::from("New"), |s| {
479            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
480                s.0, s.0,
481            ])))
482        });
483
484        let (tag, encoded) = registry.encode(&Sample(3)).expect("encode").expect("hit");
485
486        assert_eq!(tag.as_str(), "New");
487        assert_eq!(encoded.payload.as_ref(), &[3, 3]);
488        assert_eq!(registry.len(), 1);
489    }
490
491    #[rstest]
492    fn register_encoder_replaces_encoder_and_preserves_extractors() {
493        let mut registry = EncoderRegistry::new();
494        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
495            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
496                s.0,
497            ])))
498        });
499        let correlation_id = UUID4::new();
500        let causation_id = UUID4::new();
501        registry.register_headers::<Sample, _>(move |_| Headers {
502            correlation_id: Some(correlation_id),
503            causation_id: Some(causation_id),
504        });
505        let identity = UUID4::new();
506        registry.register_identity::<Sample, _>(move |_| Some(identity));
507
508        registry.register_encoder::<Sample>(
509            Ustr::from("Stateful"),
510            Arc::new(StatefulEncoder { prefix: 0xA5 }),
511        );
512        let sample = Sample(7);
513        let (payload_type, encoded) = registry.encode(&sample).expect("encode").expect("hit");
514        let headers = registry
515            .headers_for_any(&sample as &dyn Any)
516            .expect("headers");
517
518        assert_eq!(payload_type.as_str(), "Stateful");
519        assert_eq!(
520            encoded,
521            EncodedPayload::without_indices(Bytes::from_static(&[0xA5, 7])),
522        );
523        assert_eq!(
524            headers,
525            Headers {
526                correlation_id: Some(correlation_id),
527                causation_id: Some(causation_id),
528            },
529        );
530        assert_eq!(
531            registry.identity_for_any(&sample as &dyn Any),
532            Some(identity)
533        );
534        assert_eq!(registry.len(), 1);
535    }
536
537    #[rstest]
538    fn registry_is_empty_by_default() {
539        let registry = EncoderRegistry::new();
540
541        assert!(registry.is_empty());
542        assert_eq!(registry.len(), 0);
543        assert!(!registry.contains::<Other>());
544    }
545
546    #[rstest]
547    fn encode_any_dispatches_by_concrete_type_id() {
548        // The bus tap reaches the registry through `encode_any` because the static
549        // type is not in scope at the dispatch site. Verify the &dyn Any lookup
550        // resolves to the same registration as the typed `encode<T>` path.
551        let mut registry = EncoderRegistry::new();
552        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
553            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
554                s.0,
555            ])))
556        });
557
558        let sample = Sample(5);
559        let (tag, encoded) = registry
560            .encode_any(&sample as &dyn Any)
561            .expect("encode_any")
562            .expect("hit");
563
564        assert_eq!(tag.as_str(), "Sample");
565        assert_eq!(encoded.payload.as_ref(), &[5]);
566    }
567
568    #[rstest]
569    fn encode_any_returns_none_for_unregistered_type() {
570        // Out-of-allow-list messages must surface as `Ok(None)` so the adapter can
571        // skip them silently at the dispatch boundary rather than treating them as
572        // capture failures.
573        let registry = EncoderRegistry::new();
574
575        let unregistered = Other;
576        let outcome = registry
577            .encode_any(&unregistered as &dyn Any)
578            .expect("encode_any");
579
580        assert!(outcome.is_none());
581    }
582
583    #[rstest]
584    fn encoder_payload_type_override_overrides_registered_tag() {
585        // Envelope encoders (TradingCommand, OrderEventAny) need to stamp the
586        // inner-variant tag on captured entries so forensics scans see the same
587        // payload_type as the bare-type capture path. The override mechanism is
588        // tested here independent of the bus surface so the registry contract
589        // is self-documenting.
590        let mut registry = EncoderRegistry::new();
591        registry.register::<Sample, _>(Ustr::from("Wrapper"), |s| {
592            Ok(EncodedPayload::with_payload_type(
593                Ustr::from("Inner"),
594                Bytes::copy_from_slice(&[s.0]),
595                Vec::new(),
596            ))
597        });
598
599        let (tag, _) = registry.encode(&Sample(1)).expect("encode").expect("hit");
600        assert_eq!(tag.as_str(), "Inner");
601
602        let (any_tag, _) = registry
603            .encode_any(&Sample(1) as &dyn Any)
604            .expect("encode_any")
605            .expect("hit");
606        assert_eq!(any_tag.as_str(), "Inner");
607    }
608
609    #[rstest]
610    fn registered_type_without_headers_extractor_returns_empty_headers() {
611        let mut registry = EncoderRegistry::new();
612        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
613            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
614                s.0,
615            ])))
616        });
617
618        let headers = registry
619            .headers_for_any(&Sample(1) as &dyn Any)
620            .expect("hit");
621        assert_eq!(headers, Headers::empty());
622    }
623
624    #[rstest]
625    fn headers_for_any_returns_none_for_unregistered_type() {
626        let registry = EncoderRegistry::new();
627        let outcome = registry.headers_for_any(&Other as &dyn Any);
628
629        assert!(outcome.is_none());
630    }
631
632    #[rstest]
633    fn register_with_headers_uses_extractor() {
634        // The tap consults `headers_for_any` to populate the entry's Headers; the
635        // registered extractor must reach the captured message and produce the right
636        // values.
637        let mut registry = EncoderRegistry::new();
638        let causation = nautilus_core::UUID4::new();
639        let causation_captured = causation;
640        registry.register_with_headers::<Sample, _, _>(
641            Ustr::from("Sample"),
642            |s| {
643                Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
644                    s.0,
645                ])))
646            },
647            move |_| Headers {
648                correlation_id: None,
649                causation_id: Some(causation_captured),
650            },
651        );
652
653        let headers = registry
654            .headers_for_any(&Sample(1) as &dyn Any)
655            .expect("hit");
656        assert_eq!(headers.causation_id, Some(causation));
657    }
658
659    #[rstest]
660    fn register_headers_overrides_default_extractor_post_register() {
661        // Callers can attach a headers extractor after registering the encoder, which
662        // is how shared encoder helpers (default_registry) compose with site-specific
663        // header propagation.
664        let mut registry = EncoderRegistry::new();
665        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
666            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
667                s.0,
668            ])))
669        });
670        let correlation = nautilus_core::UUID4::new();
671        let correlation_captured = correlation;
672        registry.register_headers::<Sample, _>(move |_| Headers {
673            correlation_id: Some(correlation_captured),
674            causation_id: None,
675        });
676
677        let headers = registry
678            .headers_for_any(&Sample(1) as &dyn Any)
679            .expect("hit");
680        assert_eq!(headers.correlation_id, Some(correlation));
681    }
682
683    #[rstest]
684    fn identity_for_any_returns_none_without_extractor() {
685        let mut registry = EncoderRegistry::new();
686        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
687            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
688                s.0,
689            ])))
690        });
691
692        assert!(registry.identity_for_any(&Sample(1) as &dyn Any).is_none());
693        assert!(registry.identity_for_any(&Other as &dyn Any).is_none());
694    }
695
696    #[rstest]
697    fn register_identity_extracts_and_survives_re_register() {
698        // The adapter dedups on the extracted identity; the extractor must survive a
699        // subsequent encoder re-registration the same way headers extractors do.
700        let mut registry = EncoderRegistry::new();
701        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
702            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
703                s.0,
704            ])))
705        });
706        let identity = nautilus_core::UUID4::new();
707        registry.register_identity::<Sample, _>(move |_| Some(identity));
708
709        assert_eq!(
710            registry.identity_for_any(&Sample(1) as &dyn Any),
711            Some(identity),
712        );
713
714        registry.register::<Sample, _>(Ustr::from("New"), |s| {
715            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
716                s.0, s.0,
717            ])))
718        });
719
720        assert_eq!(
721            registry.identity_for_any(&Sample(1) as &dyn Any),
722            Some(identity),
723        );
724    }
725
726    #[rstest]
727    fn register_identity_for_unregistered_type_is_silent_noop() {
728        let mut registry = EncoderRegistry::new();
729        registry.register_identity::<Sample, _>(|_| Some(nautilus_core::UUID4::new()));
730
731        assert!(!registry.contains::<Sample>());
732        assert!(registry.identity_for_any(&Sample(1) as &dyn Any).is_none());
733    }
734
735    #[rstest]
736    fn register_headers_for_unregistered_type_is_silent_noop() {
737        let mut registry = EncoderRegistry::new();
738        registry.register_headers::<Sample, _>(|_| Headers::empty());
739
740        assert!(!registry.contains::<Sample>());
741        assert!(registry.headers_for_any(&Sample(1) as &dyn Any).is_none());
742    }
743
744    #[rstest]
745    fn re_registering_preserves_existing_headers_extractor() {
746        // A subsequent `register` for the same type must keep the previously-attached
747        // headers extractor so callers that compose encoder + headers in two phases do
748        // not silently lose the extractor when the encoder is replaced.
749        let mut registry = EncoderRegistry::new();
750        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
751            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
752                s.0,
753            ])))
754        });
755        let causation = nautilus_core::UUID4::new();
756        let causation_captured = causation;
757        registry.register_headers::<Sample, _>(move |_| Headers {
758            correlation_id: None,
759            causation_id: Some(causation_captured),
760        });
761        registry.register::<Sample, _>(Ustr::from("New"), |s| {
762            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
763                s.0, s.0,
764            ])))
765        });
766
767        let (tag, _) = registry.encode(&Sample(3)).expect("encode").expect("hit");
768        assert_eq!(tag.as_str(), "New");
769        let headers = registry
770            .headers_for_any(&Sample(3) as &dyn Any)
771            .expect("hit");
772        assert_eq!(headers.causation_id, Some(causation));
773    }
774}
nautilus_event_store/capture/registry.rs

nautilus_event_store/capture/
registry.rs
