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    /// encoder 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 an error if an encoder is registered but rejects the message.
351    pub fn encode<T: 'static>(
352        &self,
353        message: &T,
354    ) -> Result<Option<(PayloadType, EncodedPayload)>, EncodeError> {
355        let Some(reg) = self.by_type.get(&TypeId::of::<T>()) else {
356            return Ok(None);
357        };
358
359        let encoded = reg.encoder.encode(message as &dyn Any)?;
360        let payload_type = encoded.payload_type.unwrap_or(reg.payload_type);
361        Ok(Some((payload_type, encoded)))
362    }
363
364    /// Encodes a type-erased `message` if an encoder is registered for the concrete type.
365    ///
366    /// Mirror of [`Self::encode`] for `&dyn Any` callers; the bus tap reaches the
367    /// capture path through this entry point because dispatch returns a `&dyn Any` and
368    /// the static type is not in scope.
369    ///
370    /// # Errors
371    ///
372    /// Returns an error if an encoder is registered for the concrete type but rejects the message.
373    pub fn encode_any(
374        &self,
375        message: &dyn Any,
376    ) -> Result<Option<(PayloadType, EncodedPayload)>, EncodeError> {
377        let Some(reg) = self.by_type.get(&message.type_id()) else {
378            return Ok(None);
379        };
380
381        let encoded = reg.encoder.encode(message)?;
382        let payload_type = encoded.payload_type.unwrap_or(reg.payload_type);
383        Ok(Some((payload_type, encoded)))
384    }
385
386    /// Returns the headers carried by `message` if a registration exists for its concrete
387    /// type. Returns `None` when no encoder is registered: the adapter then drops the
388    /// message without further work.
389    #[must_use]
390    pub fn headers_for_any(&self, message: &dyn Any) -> Option<Headers> {
391        self.by_type
392            .get(&message.type_id())
393            .map(|reg| reg.headers.extract(message))
394    }
395
396    /// Returns the logical identity carried by `message` if its registration has an
397    /// identity extractor. Returns `None` when the type is unregistered or carries no
398    /// extractor; the adapter then captures per dispatch.
399    #[must_use]
400    pub fn identity_for_any(&self, message: &dyn Any) -> Option<UUID4> {
401        self.by_type
402            .get(&message.type_id())
403            .and_then(|reg| reg.identity.as_ref())
404            .and_then(|identity| identity.extract(message))
405    }
406}
407
408#[cfg(test)]
409mod tests {
410    use std::sync::Arc;
411
412    use bytes::Bytes;
413    use rstest::rstest;
414    use ustr::Ustr;
415
416    use super::*;
417
418    #[derive(Debug)]
419    struct Sample(u8);
420
421    #[derive(Debug)]
422    struct Other;
423
424    #[derive(Debug)]
425    struct StatefulEncoder {
426        prefix: u8,
427    }
428
429    impl Encode for StatefulEncoder {
430        fn encode(&self, message: &dyn Any) -> Result<EncodedPayload, EncodeError> {
431            let sample = message
432                .downcast_ref::<Sample>()
433                .ok_or(EncodeError::TypeMismatch {
434                    expected: std::any::type_name::<Sample>(),
435                })?;
436            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
437                self.prefix,
438                sample.0,
439            ])))
440        }
441    }
442
443    #[rstest]
444    fn unknown_type_returns_none() {
445        let registry = EncoderRegistry::new();
446
447        assert!(registry.encode(&Sample(1)).expect("encode").is_none());
448        assert!(!registry.contains::<Sample>());
449    }
450
451    #[rstest]
452    fn registered_type_returns_payload_type_and_payload() {
453        let mut registry = EncoderRegistry::new();
454        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
455            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
456                s.0,
457            ])))
458        });
459
460        let (tag, encoded) = registry.encode(&Sample(9)).expect("encode").expect("hit");
461
462        assert_eq!(tag.as_str(), "Sample");
463        assert_eq!(encoded.payload.as_ref(), &[9]);
464        assert!(registry.contains::<Sample>());
465        assert_eq!(registry.len(), 1);
466    }
467
468    #[rstest]
469    fn re_registering_replaces_prior_encoder() {
470        let mut registry = EncoderRegistry::new();
471        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
472            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
473                s.0,
474            ])))
475        });
476        registry.register::<Sample, _>(Ustr::from("New"), |s| {
477            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
478                s.0, s.0,
479            ])))
480        });
481
482        let (tag, encoded) = registry.encode(&Sample(3)).expect("encode").expect("hit");
483
484        assert_eq!(tag.as_str(), "New");
485        assert_eq!(encoded.payload.as_ref(), &[3, 3]);
486        assert_eq!(registry.len(), 1);
487    }
488
489    #[rstest]
490    fn register_encoder_replaces_encoder_and_preserves_extractors() {
491        let mut registry = EncoderRegistry::new();
492        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
493            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
494                s.0,
495            ])))
496        });
497        let correlation_id = UUID4::new();
498        let causation_id = UUID4::new();
499        registry.register_headers::<Sample, _>(move |_| Headers {
500            correlation_id: Some(correlation_id),
501            causation_id: Some(causation_id),
502        });
503        let identity = UUID4::new();
504        registry.register_identity::<Sample, _>(move |_| Some(identity));
505
506        registry.register_encoder::<Sample>(
507            Ustr::from("Stateful"),
508            Arc::new(StatefulEncoder { prefix: 0xA5 }),
509        );
510        let sample = Sample(7);
511        let (payload_type, encoded) = registry.encode(&sample).expect("encode").expect("hit");
512        let headers = registry
513            .headers_for_any(&sample as &dyn Any)
514            .expect("headers");
515
516        assert_eq!(payload_type.as_str(), "Stateful");
517        assert_eq!(
518            encoded,
519            EncodedPayload::without_indices(Bytes::from_static(&[0xA5, 7])),
520        );
521        assert_eq!(
522            headers,
523            Headers {
524                correlation_id: Some(correlation_id),
525                causation_id: Some(causation_id),
526            },
527        );
528        assert_eq!(
529            registry.identity_for_any(&sample as &dyn Any),
530            Some(identity)
531        );
532        assert_eq!(registry.len(), 1);
533    }
534
535    #[rstest]
536    fn registry_is_empty_by_default() {
537        let registry = EncoderRegistry::new();
538
539        assert!(registry.is_empty());
540        assert_eq!(registry.len(), 0);
541        assert!(!registry.contains::<Other>());
542    }
543
544    #[rstest]
545    fn encode_any_dispatches_by_concrete_type_id() {
546        // The bus tap reaches the registry through `encode_any` because the static
547        // type is not in scope at the dispatch site. Verify the &dyn Any lookup
548        // resolves to the same registration as the typed `encode<T>` path.
549        let mut registry = EncoderRegistry::new();
550        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
551            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
552                s.0,
553            ])))
554        });
555
556        let sample = Sample(5);
557        let (tag, encoded) = registry
558            .encode_any(&sample as &dyn Any)
559            .expect("encode_any")
560            .expect("hit");
561
562        assert_eq!(tag.as_str(), "Sample");
563        assert_eq!(encoded.payload.as_ref(), &[5]);
564    }
565
566    #[rstest]
567    fn encode_any_returns_none_for_unregistered_type() {
568        // Out-of-allow-list messages must surface as `Ok(None)` so the adapter can
569        // skip them silently at the dispatch boundary rather than treating them as
570        // capture failures.
571        let registry = EncoderRegistry::new();
572
573        let unregistered = Other;
574        let outcome = registry
575            .encode_any(&unregistered as &dyn Any)
576            .expect("encode_any");
577
578        assert!(outcome.is_none());
579    }
580
581    #[rstest]
582    fn encoder_payload_type_override_overrides_registered_tag() {
583        // Envelope encoders (TradingCommand, OrderEventAny) need to stamp the
584        // inner-variant tag on captured entries so forensics scans see the same
585        // payload_type as the bare-type capture path. The override mechanism is
586        // tested here independent of the bus surface so the registry contract
587        // is self-documenting.
588        let mut registry = EncoderRegistry::new();
589        registry.register::<Sample, _>(Ustr::from("Wrapper"), |s| {
590            Ok(EncodedPayload::with_payload_type(
591                Ustr::from("Inner"),
592                Bytes::copy_from_slice(&[s.0]),
593                Vec::new(),
594            ))
595        });
596
597        let (tag, _) = registry.encode(&Sample(1)).expect("encode").expect("hit");
598        assert_eq!(tag.as_str(), "Inner");
599
600        let (any_tag, _) = registry
601            .encode_any(&Sample(1) as &dyn Any)
602            .expect("encode_any")
603            .expect("hit");
604        assert_eq!(any_tag.as_str(), "Inner");
605    }
606
607    #[rstest]
608    fn registered_type_without_headers_extractor_returns_empty_headers() {
609        let mut registry = EncoderRegistry::new();
610        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
611            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
612                s.0,
613            ])))
614        });
615
616        let headers = registry
617            .headers_for_any(&Sample(1) as &dyn Any)
618            .expect("hit");
619        assert_eq!(headers, Headers::empty());
620    }
621
622    #[rstest]
623    fn headers_for_any_returns_none_for_unregistered_type() {
624        let registry = EncoderRegistry::new();
625        let outcome = registry.headers_for_any(&Other as &dyn Any);
626
627        assert!(outcome.is_none());
628    }
629
630    #[rstest]
631    fn register_with_headers_uses_extractor() {
632        // The tap consults `headers_for_any` to populate the entry's Headers; the
633        // registered extractor must reach the captured message and produce the right
634        // values.
635        let mut registry = EncoderRegistry::new();
636        let causation = nautilus_core::UUID4::new();
637        let causation_captured = causation;
638        registry.register_with_headers::<Sample, _, _>(
639            Ustr::from("Sample"),
640            |s| {
641                Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
642                    s.0,
643                ])))
644            },
645            move |_| Headers {
646                correlation_id: None,
647                causation_id: Some(causation_captured),
648            },
649        );
650
651        let headers = registry
652            .headers_for_any(&Sample(1) as &dyn Any)
653            .expect("hit");
654        assert_eq!(headers.causation_id, Some(causation));
655    }
656
657    #[rstest]
658    fn register_headers_overrides_default_extractor_post_register() {
659        // Callers can attach a headers extractor after registering the encoder, which
660        // is how the shared default_registry encoders compose with site-specific
661        // header propagation.
662        let mut registry = EncoderRegistry::new();
663        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
664            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
665                s.0,
666            ])))
667        });
668        let correlation = nautilus_core::UUID4::new();
669        let correlation_captured = correlation;
670        registry.register_headers::<Sample, _>(move |_| Headers {
671            correlation_id: Some(correlation_captured),
672            causation_id: None,
673        });
674
675        let headers = registry
676            .headers_for_any(&Sample(1) as &dyn Any)
677            .expect("hit");
678        assert_eq!(headers.correlation_id, Some(correlation));
679    }
680
681    #[rstest]
682    fn identity_for_any_returns_none_without_extractor() {
683        let mut registry = EncoderRegistry::new();
684        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
685            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
686                s.0,
687            ])))
688        });
689
690        assert!(registry.identity_for_any(&Sample(1) as &dyn Any).is_none());
691        assert!(registry.identity_for_any(&Other as &dyn Any).is_none());
692    }
693
694    #[rstest]
695    fn register_identity_extracts_and_survives_re_register() {
696        // The adapter dedups on the extracted identity; the extractor must survive a
697        // subsequent encoder re-registration the same way headers extractors do.
698        let mut registry = EncoderRegistry::new();
699        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
700            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
701                s.0,
702            ])))
703        });
704        let identity = nautilus_core::UUID4::new();
705        registry.register_identity::<Sample, _>(move |_| Some(identity));
706
707        assert_eq!(
708            registry.identity_for_any(&Sample(1) as &dyn Any),
709            Some(identity),
710        );
711
712        registry.register::<Sample, _>(Ustr::from("New"), |s| {
713            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
714                s.0, s.0,
715            ])))
716        });
717
718        assert_eq!(
719            registry.identity_for_any(&Sample(1) as &dyn Any),
720            Some(identity),
721        );
722    }
723
724    #[rstest]
725    fn register_identity_for_unregistered_type_is_silent_noop() {
726        let mut registry = EncoderRegistry::new();
727        registry.register_identity::<Sample, _>(|_| Some(nautilus_core::UUID4::new()));
728
729        assert!(!registry.contains::<Sample>());
730        assert!(registry.identity_for_any(&Sample(1) as &dyn Any).is_none());
731    }
732
733    #[rstest]
734    fn register_headers_for_unregistered_type_is_silent_noop() {
735        let mut registry = EncoderRegistry::new();
736        registry.register_headers::<Sample, _>(|_| Headers::empty());
737
738        assert!(!registry.contains::<Sample>());
739        assert!(registry.headers_for_any(&Sample(1) as &dyn Any).is_none());
740    }
741
742    #[rstest]
743    fn re_registering_preserves_existing_headers_extractor() {
744        // A subsequent `register` for the same type must keep the previously-attached
745        // headers extractor so callers that compose encoder + headers in two phases do
746        // not silently lose the extractor when the encoder is replaced.
747        let mut registry = EncoderRegistry::new();
748        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
749            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
750                s.0,
751            ])))
752        });
753        let causation = nautilus_core::UUID4::new();
754        let causation_captured = causation;
755        registry.register_headers::<Sample, _>(move |_| Headers {
756            correlation_id: None,
757            causation_id: Some(causation_captured),
758        });
759        registry.register::<Sample, _>(Ustr::from("New"), |s| {
760            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
761                s.0, s.0,
762            ])))
763        });
764
765        let (tag, _) = registry.encode(&Sample(3)).expect("encode").expect("hit");
766        assert_eq!(tag.as_str(), "New");
767        let headers = registry
768            .headers_for_any(&Sample(3) as &dyn Any)
769            .expect("hit");
770        assert_eq!(headers.causation_id, Some(causation));
771    }
772}
nautilus_event_store/capture/registry.rs

nautilus_event_store/capture/
registry.rs
