nautilus_model/data/
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//! Registries for custom data: JSON (de)serialization and Arrow encode/decode.
17//!
18//! Mirrors Python's `register_custom_data_class` surface in `custom.py`.
19//! The registry only stores type name -> callbacks for lookup; each type provides
20//! its own deserialize/encode/decode via the trait or registration.
21
22use std::sync::Arc;
23
24#[cfg(feature = "arrow")]
25use arrow::{
26    datatypes::{DataType as ArrowDataType, Field, Schema},
27    record_batch::RecordBatch,
28};
29use dashmap::{DashMap, mapref::entry::Entry};
30use nautilus_core::Params;
31#[cfg(feature = "python")]
32use pyo3::types::PyAnyMethods;
33
34use crate::data::{CustomData, CustomDataTrait, Data, DataType};
35
36pub type JsonDeserializer =
37    Box<dyn Fn(serde_json::Value) -> Result<Arc<dyn CustomDataTrait>, anyhow::Error> + Send + Sync>;
38#[cfg(feature = "arrow")]
39pub type ArrowEncoder =
40    Box<dyn Fn(&[Arc<dyn CustomDataTrait>]) -> Result<RecordBatch, anyhow::Error> + Send + Sync>;
41#[cfg(feature = "arrow")]
42pub type ArrowDecoder = Box<
43    dyn Fn(
44            &std::collections::HashMap<String, String>,
45            RecordBatch,
46        ) -> Result<Vec<Data>, anyhow::Error>
47        + Send
48        + Sync,
49>;
50
51/// Validates that a custom Arrow write schema contains no unsupported opaque byte fields.
52///
53/// `allow_binary` is reserved for the Rust macro's documented `Vec<u8>` exemption. Python
54/// schemas cannot prove that provenance and must pass `false`.
55///
56/// # Errors
57///
58/// Returns an error naming the first opaque byte field.
59#[cfg(feature = "arrow")]
60pub fn validate_custom_arrow_schema(
61    type_name: &str,
62    schema: &Schema,
63    allow_binary: bool,
64) -> anyhow::Result<()> {
65    for field in schema.fields() {
66        validate_custom_arrow_field(type_name, field, allow_binary)?;
67    }
68    Ok(())
69}
70
71#[cfg(feature = "arrow")]
72fn validate_custom_arrow_field(
73    type_name: &str,
74    field: &Field,
75    allow_binary: bool,
76) -> anyhow::Result<()> {
77    match field.data_type() {
78        ArrowDataType::Binary if allow_binary => Ok(()),
79        ArrowDataType::Binary
80        | ArrowDataType::LargeBinary
81        | ArrowDataType::BinaryView
82        | ArrowDataType::FixedSizeBinary(_) => anyhow::bail!(
83            "custom write schema `{type_name}` contains opaque byte field `{}`: {}",
84            field.name(),
85            field.data_type(),
86        ),
87        ArrowDataType::List(child)
88        | ArrowDataType::LargeList(child)
89        | ArrowDataType::ListView(child)
90        | ArrowDataType::LargeListView(child)
91        | ArrowDataType::FixedSizeList(child, _)
92        | ArrowDataType::Map(child, _) => {
93            validate_custom_arrow_field(type_name, child, allow_binary)
94        }
95        ArrowDataType::Struct(children) => {
96            for child in children {
97                validate_custom_arrow_field(type_name, child, allow_binary)?;
98            }
99            Ok(())
100        }
101        ArrowDataType::Dictionary(_, value) => {
102            let child = Field::new(field.name(), value.as_ref().clone(), field.is_nullable());
103            validate_custom_arrow_field(type_name, &child, allow_binary)
104        }
105        _ => Ok(()),
106    }
107}
108
109struct Registries {
110    json: DashMap<String, JsonDeserializer>,
111    #[cfg(feature = "arrow")]
112    arrow: DashMap<String, (Arc<Schema>, ArrowEncoder, ArrowDecoder)>,
113}
114
115fn registries() -> &'static Registries {
116    static REGISTRIES: std::sync::OnceLock<Registries> = std::sync::OnceLock::new();
117    REGISTRIES.get_or_init(|| Registries {
118        json: DashMap::new(),
119        #[cfg(feature = "arrow")]
120        arrow: DashMap::new(),
121    })
122}
123
124/// Registers a JSON deserializer for the given custom data type name.
125/// When `Data::deserialize` sees this type name, it will call this function.
126///
127/// # Errors
128/// Returns an error if the type is already registered.
129pub fn register_json_deserializer(
130    type_name: &str,
131    deserializer: JsonDeserializer,
132) -> Result<(), anyhow::Error> {
133    let reg = registries();
134    match reg.json.entry(type_name.to_string()) {
135        Entry::Occupied(_) => {
136            anyhow::bail!("Custom data type \"{type_name}\" is already registered for JSON");
137        }
138        Entry::Vacant(v) => {
139            v.insert(deserializer);
140            Ok(())
141        }
142    }
143}
144
145/// Registers a JSON deserializer for the given custom data type name if not already registered.
146/// If the type is already registered, returns `Ok(())` without overwriting (idempotent).
147/// Use this where repeated registration can occur (e.g. module init).
148///
149/// # Errors
150/// Does not return an error (idempotent insert into `DashMap`).
151pub fn ensure_json_deserializer_registered(
152    type_name: &str,
153    deserializer: JsonDeserializer,
154) -> Result<(), anyhow::Error> {
155    let reg = registries();
156    reg.json
157        .entry(type_name.to_string())
158        .or_insert_with(|| deserializer);
159    Ok(())
160}
161
162/// Parses a "`data_type`" JSON object into `DataType` (`type_name`, metadata, identifier).
163fn parse_data_type_from_value(value: &serde_json::Value) -> Option<DataType> {
164    let obj = value.get("data_type")?.as_object()?;
165    let type_name = obj.get("type_name")?.as_str()?;
166    let metadata = obj.get("metadata").and_then(|m| {
167        if m.is_null() {
168            None
169        } else {
170            let p: Params = serde_json::from_value(m.clone()).ok()?;
171            if p.is_empty() { None } else { Some(p) }
172        }
173    });
174    let identifier = obj
175        .get("identifier")
176        .and_then(|v| v.as_str())
177        .map(String::from);
178    Some(DataType::new(type_name, metadata, identifier))
179}
180
181/// Parses the canonical `CustomData` JSON envelope `{ type, data_type, payload }` and returns
182/// the payload value to pass to the registered type deserializer. Does not depend on
183/// user payload field names.
184fn parse_envelope_payload(value: &serde_json::Value) -> Result<serde_json::Value, anyhow::Error> {
185    let payload = value
186        .get("payload")
187        .cloned()
188        .ok_or_else(|| anyhow::anyhow!("CustomData JSON missing 'payload' field"))?;
189    Ok(payload)
190}
191
192/// Looks up and runs the JSON deserializer for the given type name.
193/// Returns `None` if the type is not registered.
194///
195/// # Errors
196/// Returns an error if the deserializer fails.
197pub fn deserialize_custom_from_json(
198    type_name: &str,
199    value: &serde_json::Value,
200) -> Result<Option<Data>, anyhow::Error> {
201    let reg = registries();
202    let deserializer_ref = match reg.json.get(type_name) {
203        Some(d) => d,
204        None => return Ok(None),
205    };
206    let data_type = parse_data_type_from_value(value);
207    let payload = parse_envelope_payload(value)?;
208    let arc = deserializer_ref.value()(payload)?;
209    let custom = match data_type {
210        Some(dt) => CustomData::new(arc, dt),
211        None => CustomData::from_arc(arc),
212    };
213    Ok(Some(Data::Custom(custom)))
214}
215
216/// Registers Arrow schema, encoder, and decoder for the given custom data type name.
217///
218/// # Errors
219/// Returns an error if the type is already registered for Arrow.
220#[cfg(feature = "arrow")]
221pub fn register_arrow(
222    type_name: &str,
223    schema: Arc<Schema>,
224    encoder: ArrowEncoder,
225    decoder: ArrowDecoder,
226) -> Result<(), anyhow::Error> {
227    let reg = registries();
228    match reg.arrow.entry(type_name.to_string()) {
229        Entry::Occupied(_) => {
230            anyhow::bail!("Custom data type \"{type_name}\" is already registered for Arrow");
231        }
232        Entry::Vacant(v) => {
233            v.insert((schema, encoder, decoder));
234            Ok(())
235        }
236    }
237}
238
239/// Registers Arrow schema, encoder, and decoder for the given custom data type name if not already
240/// registered. If the type is already registered, returns `Ok(())` without overwriting (idempotent).
241/// Use this where repeated registration can occur (e.g. module init).
242///
243/// # Errors
244/// Does not return an error (idempotent insert into `DashMap`).
245#[cfg(feature = "arrow")]
246pub fn ensure_arrow_registered(
247    type_name: &str,
248    schema: Arc<Schema>,
249    encoder: ArrowEncoder,
250    decoder: ArrowDecoder,
251) -> Result<(), anyhow::Error> {
252    let reg = registries();
253    reg.arrow
254        .entry(type_name.to_string())
255        .or_insert_with(|| (schema, encoder, decoder));
256    Ok(())
257}
258
259/// Returns the Arrow schema for the given custom type name, if registered.
260#[must_use]
261#[cfg(feature = "arrow")]
262pub fn get_arrow_schema(type_name: &str) -> Option<Arc<Schema>> {
263    let reg = registries();
264    reg.arrow
265        .get(type_name)
266        .map(|entry| Arc::clone(&entry.value().0))
267}
268
269/// Encodes a slice of custom data trait objects to a `RecordBatch` using the registered encoder.
270///
271/// # Errors
272/// Returns an error if the type is not registered or encoding fails.
273#[cfg(feature = "arrow")]
274pub fn encode_custom_to_arrow(
275    type_name: &str,
276    items: &[Arc<dyn CustomDataTrait>],
277) -> Result<Option<RecordBatch>, anyhow::Error> {
278    let reg = registries();
279    let entry = match reg.arrow.get(type_name) {
280        Some(e) => e,
281        None => return Ok(None),
282    };
283    let encoder = &entry.value().1;
284    encoder(items).map(Some)
285}
286
287/// Decodes a `RecordBatch` into `Vec<Data>` using the registered decoder.
288///
289/// # Errors
290/// Returns an error if the type is not registered or decoding fails.
291#[expect(
292    clippy::implicit_hasher,
293    reason = "callers always use the default hasher"
294)]
295#[cfg(feature = "arrow")]
296pub fn decode_custom_from_arrow(
297    type_name: &str,
298    metadata: &std::collections::HashMap<String, String>,
299    record_batch: RecordBatch,
300) -> Result<Option<Vec<Data>>, anyhow::Error> {
301    let reg = registries();
302    let entry = match reg.arrow.get(type_name) {
303        Some(e) => e,
304        None => return Ok(None),
305    };
306    let decoder = &entry.value().2;
307    decoder(metadata, record_batch).map(Some)
308}
309
310#[cfg(feature = "python")]
311pub type PyExtractor = Box<
312    dyn for<'a> Fn(&pyo3::Bound<'a, pyo3::PyAny>) -> Option<Arc<dyn CustomDataTrait>> + Send + Sync,
313>;
314
315#[cfg(feature = "python")]
316fn py_extractors() -> &'static DashMap<String, PyExtractor> {
317    static PY_EXTRACTORS: std::sync::OnceLock<DashMap<String, PyExtractor>> =
318        std::sync::OnceLock::new();
319    PY_EXTRACTORS.get_or_init(DashMap::new)
320}
321
322/// Registers a `PyExtractor` for the given custom data type name.
323/// Used by `CustomData` constructor to convert Python objects to `Arc<dyn CustomDataTrait>`.
324///
325/// # Errors
326/// Returns an error if the type is already registered.
327#[cfg(feature = "python")]
328pub fn register_py_extractor(type_name: &str, extractor: PyExtractor) -> Result<(), anyhow::Error> {
329    let reg = py_extractors();
330    match reg.entry(type_name.to_string()) {
331        Entry::Occupied(_) => {
332            anyhow::bail!(
333                "Custom data type \"{type_name}\" is already registered for Python extraction"
334            );
335        }
336        Entry::Vacant(v) => {
337            v.insert(extractor);
338            Ok(())
339        }
340    }
341}
342
343/// Registers a `PyExtractor` for the given custom data type name if not already registered.
344/// If the type is already registered, returns `Ok(())` without overwriting (idempotent).
345/// Use this where repeated registration can occur (e.g. module init).
346///
347/// # Errors
348/// Does not return an error (idempotent insert into `DashMap`).
349#[cfg(feature = "python")]
350pub fn ensure_py_extractor_registered(
351    type_name: &str,
352    extractor: PyExtractor,
353) -> Result<(), anyhow::Error> {
354    let reg = py_extractors();
355    reg.entry(type_name.to_string())
356        .or_insert_with(|| extractor);
357    Ok(())
358}
359
360/// Tries to extract `Arc<dyn CustomDataTrait>` from a Python object using the registered extractor.
361/// Returns None if no extractor is registered or extraction fails.
362#[cfg(feature = "python")]
363#[must_use]
364pub fn try_extract_from_py(
365    type_name: &str,
366    obj: &pyo3::Bound<'_, pyo3::PyAny>,
367) -> Option<Arc<dyn CustomDataTrait>> {
368    let reg = py_extractors();
369    let entry = reg.get(type_name)?;
370    let extractor = entry.value();
371    extractor(obj)
372}
373
374#[cfg(feature = "python")]
375type RustExtractorFactory = Box<dyn Fn() -> PyExtractor + Send + Sync>;
376
377#[cfg(feature = "python")]
378fn rust_extractor_factories() -> &'static DashMap<String, RustExtractorFactory> {
379    static RUST_EXTRACTOR_FACTORIES: std::sync::OnceLock<DashMap<String, RustExtractorFactory>> =
380        std::sync::OnceLock::new();
381    RUST_EXTRACTOR_FACTORIES.get_or_init(DashMap::new)
382}
383
384/// Registers a factory that produces a `PyExtractor` for the given type name.
385/// Crates (e.g. persistence) call this at load time for each Rust custom data type.
386/// When `register_custom_data_class(cls)` is called with that type's class, the factory is invoked
387/// and the extractor is registered in the main `PyExtractor` registry.
388///
389/// # Errors
390/// Returns an error if the type name is already registered.
391#[cfg(feature = "python")]
392pub fn register_rust_extractor_factory(
393    type_name: &str,
394    factory: RustExtractorFactory,
395) -> Result<(), anyhow::Error> {
396    let reg = rust_extractor_factories();
397    match reg.entry(type_name.to_string()) {
398        Entry::Occupied(_) => {
399            anyhow::bail!("Rust extractor factory for \"{type_name}\" is already registered");
400        }
401        Entry::Vacant(v) => {
402            v.insert(factory);
403            Ok(())
404        }
405    }
406}
407
408/// Registers a factory that produces a `PyExtractor` for the given type name if not already
409/// registered. If the type is already registered, returns `Ok(())` without overwriting (idempotent).
410/// Use this where repeated registration can occur (e.g. module load).
411///
412/// # Errors
413/// Does not return an error (idempotent insert into `DashMap`).
414#[cfg(feature = "python")]
415pub fn ensure_rust_extractor_factory_registered(
416    type_name: &str,
417    factory: RustExtractorFactory,
418) -> Result<(), anyhow::Error> {
419    let reg = rust_extractor_factories();
420    reg.entry(type_name.to_string()).or_insert_with(|| factory);
421    Ok(())
422}
423
424/// Registers a Rust custom data type for Python extraction. Call once per type at module load
425/// (e.g. in the persistence PyO3 module). Uses [`register_rust_extractor_factory`] with a
426/// factory that builds the extractor for `T`.
427///
428/// # Errors
429/// Returns an error if the type name is already registered.
430#[cfg(feature = "python")]
431pub fn register_rust_extractor<T>() -> Result<(), anyhow::Error>
432where
433    T: CustomDataTrait + for<'a, 'py> pyo3::FromPyObject<'a, 'py> + Send + Sync + 'static,
434{
435    let type_name = T::type_name_static();
436    let factory: RustExtractorFactory = Box::new(|| {
437        Box::new(|obj: &pyo3::Bound<'_, pyo3::PyAny>| {
438            obj.extract::<T>()
439                .ok()
440                .map(|x| Arc::new(x) as Arc<dyn CustomDataTrait>)
441        })
442    });
443    register_rust_extractor_factory(type_name, factory)
444}
445
446/// Registers a Rust custom data type for Python extraction if not already registered.
447/// If the type is already registered, returns `Ok(())` without overwriting (idempotent).
448/// Use this where repeated registration can occur (e.g. module load).
449///
450/// # Errors
451/// Does not return an error (idempotent insert into `DashMap`).
452#[cfg(feature = "python")]
453pub fn ensure_rust_extractor_registered<T>() -> Result<(), anyhow::Error>
454where
455    T: CustomDataTrait + for<'a, 'py> pyo3::FromPyObject<'a, 'py> + Send + Sync + 'static,
456{
457    let type_name = T::type_name_static();
458    let factory: RustExtractorFactory = Box::new(|| {
459        Box::new(|obj: &pyo3::Bound<'_, pyo3::PyAny>| {
460            obj.extract::<T>()
461                .ok()
462                .map(|x| Arc::new(x) as Arc<dyn CustomDataTrait>)
463        })
464    });
465    ensure_rust_extractor_factory_registered(type_name, factory)
466}
467
468/// Calls the registered factory for the given type name and returns the extractor, if any.
469#[cfg(feature = "python")]
470#[must_use]
471pub fn get_rust_extractor(type_name: &str) -> Option<PyExtractor> {
472    let reg = rust_extractor_factories();
473    let factory_ref = reg.get(type_name)?;
474    Some(factory_ref.value()())
475}
476
477#[cfg(test)]
478mod tests {
479    use std::hash::{DefaultHasher, Hash, Hasher};
480
481    use nautilus_core::UnixNanos;
482    use rstest::rstest;
483    use serde::{Deserialize, Serialize};
484
485    use super::*;
486    use crate::data::{CustomData, custom::register_custom_data_json};
487
488    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
489    struct TestRegCustomData {
490        ts_init: UnixNanos,
491    }
492
493    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
494    #[serde(deny_unknown_fields)]
495    struct StrictRegCustomData {
496        ts_init: UnixNanos,
497    }
498
499    impl crate::data::HasTsInit for TestRegCustomData {
500        fn ts_init(&self) -> UnixNanos {
501            self.ts_init
502        }
503    }
504
505    impl crate::data::custom::CustomDataTrait for TestRegCustomData {
506        fn type_name(&self) -> &'static str {
507            "TestRegCustomData"
508        }
509        fn type_name_static() -> &'static str {
510            "TestRegCustomData"
511        }
512        fn as_any(&self) -> &dyn std::any::Any {
513            self
514        }
515        fn ts_event(&self) -> nautilus_core::UnixNanos {
516            self.ts_init
517        }
518        fn to_json(&self) -> anyhow::Result<String> {
519            Ok(serde_json::to_string(self)?)
520        }
521        fn clone_arc(&self) -> Arc<dyn crate::data::CustomDataTrait> {
522            Arc::new(self.clone())
523        }
524        fn eq_arc(&self, other: &dyn crate::data::CustomDataTrait) -> bool {
525            other.as_any().downcast_ref::<Self>() == Some(self)
526        }
527        fn from_json(
528            value: serde_json::Value,
529        ) -> anyhow::Result<Arc<dyn crate::data::CustomDataTrait>> {
530            let t: Self = serde_json::from_value(value)?;
531            Ok(Arc::new(t))
532        }
533    }
534
535    impl crate::data::HasTsInit for StrictRegCustomData {
536        fn ts_init(&self) -> UnixNanos {
537            self.ts_init
538        }
539    }
540
541    impl crate::data::custom::CustomDataTrait for StrictRegCustomData {
542        fn type_name(&self) -> &'static str {
543            "StrictRegCustomData"
544        }
545        fn type_name_static() -> &'static str {
546            "StrictRegCustomData"
547        }
548        fn as_any(&self) -> &dyn std::any::Any {
549            self
550        }
551        fn ts_event(&self) -> nautilus_core::UnixNanos {
552            self.ts_init
553        }
554        fn to_json(&self) -> anyhow::Result<String> {
555            Ok(serde_json::to_string(self)?)
556        }
557        fn clone_arc(&self) -> Arc<dyn crate::data::CustomDataTrait> {
558            Arc::new(self.clone())
559        }
560        fn eq_arc(&self, other: &dyn crate::data::CustomDataTrait) -> bool {
561            other.as_any().downcast_ref::<Self>() == Some(self)
562        }
563        fn from_json(
564            value: serde_json::Value,
565        ) -> anyhow::Result<Arc<dyn crate::data::CustomDataTrait>> {
566            let t: Self = serde_json::from_value(value)?;
567            Ok(Arc::new(t))
568        }
569    }
570
571    #[rstest]
572    fn json_registry_roundtrip() {
573        let _ = register_custom_data_json::<TestRegCustomData>();
574
575        let data = Data::Custom(CustomData::from_arc(Arc::new(TestRegCustomData {
576            ts_init: UnixNanos::from(100),
577        })));
578
579        let json = serde_json::to_string(&data).unwrap();
580        let back: Data = serde_json::from_str(&json).unwrap();
581
582        match (&data, &back) {
583            (Data::Custom(a), Data::Custom(b)) => {
584                assert_eq!(a.data.type_name(), b.data.type_name());
585                assert_eq!(a.data.ts_init(), b.data.ts_init());
586            }
587            _ => panic!("expected Custom variant"),
588        }
589    }
590
591    #[rstest]
592    fn json_registry_roundtrip_with_deny_unknown_fields() {
593        let _ = register_custom_data_json::<StrictRegCustomData>();
594
595        let data = Data::Custom(CustomData::from_arc(Arc::new(StrictRegCustomData {
596            ts_init: UnixNanos::from(200),
597        })));
598
599        let json = serde_json::to_string(&data).unwrap();
600        let back: Data = serde_json::from_str(&json).unwrap();
601
602        match (&data, &back) {
603            (Data::Custom(a), Data::Custom(b)) => {
604                assert_eq!(a.data.type_name(), b.data.type_name());
605                assert_eq!(a.data.ts_init(), b.data.ts_init());
606            }
607            _ => panic!("expected Custom variant"),
608        }
609    }
610
611    #[rstest]
612    fn data_type_registry_result_hashes_like_equal_values_from_all_routes() {
613        fn hash_data_type(data_type: &DataType) -> u64 {
614            let mut hasher = DefaultHasher::new();
615            data_type.hash(&mut hasher);
616            hasher.finish()
617        }
618
619        let metadata = serde_json::json!({"key": "value"});
620        let constructed = DataType::new(
621            "ExampleType",
622            Some(serde_json::from_value(metadata.clone()).unwrap()),
623            Some("catalog/path".to_string()),
624        );
625        let persistence_json = serde_json::json!({
626            "type_name": constructed.type_name(),
627            "metadata": metadata,
628            "identifier": constructed.identifier(),
629        });
630        let persisted = DataType::from_persistence_json(&persistence_json.to_string()).unwrap();
631        let registry_envelope = serde_json::json!({"data_type": persistence_json});
632        let registered = parse_data_type_from_value(&registry_envelope).unwrap();
633        let deserialization_payload = serde_json::json!({
634            "type_name": constructed.type_name(),
635            "metadata": constructed.metadata(),
636            "topic": constructed.topic(),
637            "hash": constructed.precomputed_hash() ^ u64::MAX,
638            "identifier": constructed.identifier(),
639        });
640        let deserialized: DataType = serde_json::from_value(deserialization_payload).unwrap();
641
642        // Eq and Hash both observe `topic` alone, so they cannot detect a route that drops or
643        // rewrites the other fields. Assert those explicitly as well, or a registry regression
644        // that lost `identifier` would still satisfy this test.
645        for data_type in [&deserialized, &persisted, &registered] {
646            assert_eq!(data_type, &constructed);
647            assert_eq!(hash_data_type(data_type), hash_data_type(&constructed));
648            assert_eq!(data_type.type_name(), constructed.type_name());
649            assert_eq!(data_type.metadata(), constructed.metadata());
650            assert_eq!(data_type.identifier(), constructed.identifier());
651            assert_eq!(data_type.topic(), constructed.topic());
652        }
653    }
654
655    #[rstest]
656    fn ensure_json_deserializer_registered_is_idempotent() {
657        let deserializer: JsonDeserializer = Box::new(|value| {
658            let t: TestRegCustomData = serde_json::from_value(value)?;
659            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
660        });
661        let r1 = ensure_json_deserializer_registered("IdempotentTestJson", deserializer);
662        assert!(r1.is_ok(), "first registration should succeed");
663        let deserializer2: JsonDeserializer = Box::new(|value| {
664            let t: TestRegCustomData = serde_json::from_value(value)?;
665            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
666        });
667        let r2 = ensure_json_deserializer_registered("IdempotentTestJson", deserializer2);
668        assert!(
669            r2.is_ok(),
670            "second registration with same type_name should succeed (idempotent)"
671        );
672    }
673
674    #[rstest]
675    fn register_json_deserializer_fails_on_duplicate() {
676        let deserializer: JsonDeserializer = Box::new(|value| {
677            let t: TestRegCustomData = serde_json::from_value(value)?;
678            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
679        });
680        let r1 = register_json_deserializer("StrictDuplicateTestJson", deserializer);
681        assert!(r1.is_ok());
682        let deserializer2: JsonDeserializer = Box::new(|value| {
683            let t: TestRegCustomData = serde_json::from_value(value)?;
684            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
685        });
686        let r2 = register_json_deserializer("StrictDuplicateTestJson", deserializer2);
687        assert!(r2.is_err());
688        let err_msg = r2.unwrap_err().to_string();
689        assert!(
690            err_msg.contains("already registered"),
691            "expected 'already registered' in error, found: {err_msg}"
692        );
693    }
694
695    #[rstest]
696    #[cfg(feature = "arrow")]
697    fn ensure_arrow_registered_is_idempotent() {
698        let schema = Arc::new(arrow::datatypes::Schema::empty());
699        let encoder: ArrowEncoder = Box::new(|_| {
700            Ok(arrow::record_batch::RecordBatch::new_empty(Arc::new(
701                arrow::datatypes::Schema::empty(),
702            )))
703        });
704        let decoder: ArrowDecoder = Box::new(|_, _| Ok(Vec::new()));
705
706        let r1 = ensure_arrow_registered("IdempotentTestArrow", schema, encoder, decoder);
707        assert!(r1.is_ok(), "first Arrow registration should succeed");
708
709        let schema2 = Arc::new(arrow::datatypes::Schema::empty());
710        let encoder2: ArrowEncoder = Box::new(|_| {
711            Ok(arrow::record_batch::RecordBatch::new_empty(Arc::new(
712                arrow::datatypes::Schema::empty(),
713            )))
714        });
715        let decoder2: ArrowDecoder = Box::new(|_, _| Ok(Vec::new()));
716
717        let r2 = ensure_arrow_registered("IdempotentTestArrow", schema2, encoder2, decoder2);
718        assert!(
719            r2.is_ok(),
720            "second Arrow registration with same type_name should be idempotent"
721        );
722    }
723
724    #[rstest]
725    #[cfg(feature = "arrow")]
726    fn python_custom_arrow_schema_rejects_opaque_bytes() {
727        let schema = Schema::new(vec![Field::new("payload", ArrowDataType::Binary, false)]);
728
729        let error = validate_custom_arrow_schema("PythonPayload", &schema, false).unwrap_err();
730
731        assert_eq!(
732            error.to_string(),
733            "custom write schema `PythonPayload` contains opaque byte field `payload`: Binary",
734        );
735    }
736}
nautilus_model/data/registry.rs

nautilus_model/data/
registry.rs
