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 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    #[rstest]
425    fn unknown_type_returns_none() {
426        let registry = EncoderRegistry::new();
427
428        assert!(registry.encode(&Sample(1)).expect("encode").is_none());
429        assert!(!registry.contains::<Sample>());
430    }
431
432    #[rstest]
433    fn registered_type_returns_payload_type_and_payload() {
434        let mut registry = EncoderRegistry::new();
435        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
436            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
437                s.0,
438            ])))
439        });
440
441        let (tag, encoded) = registry.encode(&Sample(9)).expect("encode").expect("hit");
442
443        assert_eq!(tag.as_str(), "Sample");
444        assert_eq!(encoded.payload.as_ref(), &[9]);
445        assert!(registry.contains::<Sample>());
446        assert_eq!(registry.len(), 1);
447    }
448
449    #[rstest]
450    fn re_registering_replaces_prior_encoder() {
451        let mut registry = EncoderRegistry::new();
452        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
453            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
454                s.0,
455            ])))
456        });
457        registry.register::<Sample, _>(Ustr::from("New"), |s| {
458            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
459                s.0, s.0,
460            ])))
461        });
462
463        let (tag, encoded) = registry.encode(&Sample(3)).expect("encode").expect("hit");
464
465        assert_eq!(tag.as_str(), "New");
466        assert_eq!(encoded.payload.as_ref(), &[3, 3]);
467        assert_eq!(registry.len(), 1);
468    }
469
470    #[rstest]
471    fn registry_is_empty_by_default() {
472        let registry = EncoderRegistry::new();
473
474        assert!(registry.is_empty());
475        assert_eq!(registry.len(), 0);
476        assert!(!registry.contains::<Other>());
477    }
478
479    #[rstest]
480    fn encode_any_dispatches_by_concrete_type_id() {
481        // The bus tap reaches the registry through `encode_any` because the static
482        // type is not in scope at the dispatch site. Verify the &dyn Any lookup
483        // resolves to the same registration as the typed `encode<T>` path.
484        let mut registry = EncoderRegistry::new();
485        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
486            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
487                s.0,
488            ])))
489        });
490
491        let sample = Sample(5);
492        let (tag, encoded) = registry
493            .encode_any(&sample as &dyn Any)
494            .expect("encode_any")
495            .expect("hit");
496
497        assert_eq!(tag.as_str(), "Sample");
498        assert_eq!(encoded.payload.as_ref(), &[5]);
499    }
500
501    #[rstest]
502    fn encode_any_returns_none_for_unregistered_type() {
503        // Out-of-allow-list messages must surface as `Ok(None)` so the adapter can
504        // skip them silently at the dispatch boundary rather than treating them as
505        // capture failures.
506        let registry = EncoderRegistry::new();
507
508        let unregistered = Other;
509        let outcome = registry
510            .encode_any(&unregistered as &dyn Any)
511            .expect("encode_any");
512
513        assert!(outcome.is_none());
514    }
515
516    #[rstest]
517    fn encoder_payload_type_override_overrides_registered_tag() {
518        // Envelope encoders (TradingCommand, OrderEventAny) need to stamp the
519        // inner-variant tag on captured entries so forensics scans see the same
520        // payload_type as the bare-type capture path. The override mechanism is
521        // tested here independent of the bus surface so the registry contract
522        // is self-documenting.
523        let mut registry = EncoderRegistry::new();
524        registry.register::<Sample, _>(Ustr::from("Wrapper"), |s| {
525            Ok(EncodedPayload::with_payload_type(
526                Ustr::from("Inner"),
527                Bytes::copy_from_slice(&[s.0]),
528                Vec::new(),
529            ))
530        });
531
532        let (tag, _) = registry.encode(&Sample(1)).expect("encode").expect("hit");
533        assert_eq!(tag.as_str(), "Inner");
534
535        let (any_tag, _) = registry
536            .encode_any(&Sample(1) as &dyn Any)
537            .expect("encode_any")
538            .expect("hit");
539        assert_eq!(any_tag.as_str(), "Inner");
540    }
541
542    #[rstest]
543    fn registered_type_without_headers_extractor_returns_empty_headers() {
544        let mut registry = EncoderRegistry::new();
545        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
546            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
547                s.0,
548            ])))
549        });
550
551        let headers = registry
552            .headers_for_any(&Sample(1) as &dyn Any)
553            .expect("hit");
554        assert_eq!(headers, Headers::empty());
555    }
556
557    #[rstest]
558    fn headers_for_any_returns_none_for_unregistered_type() {
559        let registry = EncoderRegistry::new();
560        let outcome = registry.headers_for_any(&Other as &dyn Any);
561
562        assert!(outcome.is_none());
563    }
564
565    #[rstest]
566    fn register_with_headers_uses_extractor() {
567        // The tap consults `headers_for_any` to populate the entry's Headers; the
568        // registered extractor must reach the captured message and produce the right
569        // values.
570        let mut registry = EncoderRegistry::new();
571        let causation = nautilus_core::UUID4::new();
572        let causation_captured = causation;
573        registry.register_with_headers::<Sample, _, _>(
574            Ustr::from("Sample"),
575            |s| {
576                Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
577                    s.0,
578                ])))
579            },
580            move |_| Headers {
581                correlation_id: None,
582                causation_id: Some(causation_captured),
583            },
584        );
585
586        let headers = registry
587            .headers_for_any(&Sample(1) as &dyn Any)
588            .expect("hit");
589        assert_eq!(headers.causation_id, Some(causation));
590    }
591
592    #[rstest]
593    fn register_headers_overrides_default_extractor_post_register() {
594        // Callers can attach a headers extractor after registering the encoder, which
595        // is how shared encoder helpers (default_registry) compose with site-specific
596        // header propagation.
597        let mut registry = EncoderRegistry::new();
598        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
599            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
600                s.0,
601            ])))
602        });
603        let correlation = nautilus_core::UUID4::new();
604        let correlation_captured = correlation;
605        registry.register_headers::<Sample, _>(move |_| Headers {
606            correlation_id: Some(correlation_captured),
607            causation_id: None,
608        });
609
610        let headers = registry
611            .headers_for_any(&Sample(1) as &dyn Any)
612            .expect("hit");
613        assert_eq!(headers.correlation_id, Some(correlation));
614    }
615
616    #[rstest]
617    fn identity_for_any_returns_none_without_extractor() {
618        let mut registry = EncoderRegistry::new();
619        registry.register::<Sample, _>(Ustr::from("Sample"), |s| {
620            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
621                s.0,
622            ])))
623        });
624
625        assert!(registry.identity_for_any(&Sample(1) as &dyn Any).is_none());
626        assert!(registry.identity_for_any(&Other as &dyn Any).is_none());
627    }
628
629    #[rstest]
630    fn register_identity_extracts_and_survives_re_register() {
631        // The adapter dedups on the extracted identity; the extractor must survive a
632        // subsequent encoder re-registration the same way headers extractors do.
633        let mut registry = EncoderRegistry::new();
634        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
635            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
636                s.0,
637            ])))
638        });
639        let identity = nautilus_core::UUID4::new();
640        registry.register_identity::<Sample, _>(move |_| Some(identity));
641
642        assert_eq!(
643            registry.identity_for_any(&Sample(1) as &dyn Any),
644            Some(identity),
645        );
646
647        registry.register::<Sample, _>(Ustr::from("New"), |s| {
648            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
649                s.0, s.0,
650            ])))
651        });
652
653        assert_eq!(
654            registry.identity_for_any(&Sample(1) as &dyn Any),
655            Some(identity),
656        );
657    }
658
659    #[rstest]
660    fn register_identity_for_unregistered_type_is_silent_noop() {
661        let mut registry = EncoderRegistry::new();
662        registry.register_identity::<Sample, _>(|_| Some(nautilus_core::UUID4::new()));
663
664        assert!(!registry.contains::<Sample>());
665        assert!(registry.identity_for_any(&Sample(1) as &dyn Any).is_none());
666    }
667
668    #[rstest]
669    fn register_headers_for_unregistered_type_is_silent_noop() {
670        let mut registry = EncoderRegistry::new();
671        registry.register_headers::<Sample, _>(|_| Headers::empty());
672
673        assert!(!registry.contains::<Sample>());
674        assert!(registry.headers_for_any(&Sample(1) as &dyn Any).is_none());
675    }
676
677    #[rstest]
678    fn re_registering_preserves_existing_headers_extractor() {
679        // A subsequent `register` for the same type must keep the previously-attached
680        // headers extractor so callers that compose encoder + headers in two phases do
681        // not silently lose the extractor when the encoder is replaced.
682        let mut registry = EncoderRegistry::new();
683        registry.register::<Sample, _>(Ustr::from("Old"), |s| {
684            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
685                s.0,
686            ])))
687        });
688        let causation = nautilus_core::UUID4::new();
689        let causation_captured = causation;
690        registry.register_headers::<Sample, _>(move |_| Headers {
691            correlation_id: None,
692            causation_id: Some(causation_captured),
693        });
694        registry.register::<Sample, _>(Ustr::from("New"), |s| {
695            Ok(EncodedPayload::without_indices(Bytes::copy_from_slice(&[
696                s.0, s.0,
697            ])))
698        });
699
700        let (tag, _) = registry.encode(&Sample(3)).expect("encode").expect("hit");
701        assert_eq!(tag.as_str(), "New");
702        let headers = registry
703            .headers_for_any(&Sample(3) as &dyn Any)
704            .expect("hit");
705        assert_eq!(headers.causation_id, Some(causation));
706    }
707}
nautilus_event_store/capture/registry.rs

nautilus_event_store/capture/
registry.rs
