laminar_connectors/
registry.rs

1//! Registry of connector factories, keyed by connector type string.
2
3use std::collections::{BTreeMap, HashMap};
4use std::sync::Arc;
5
6use arrow_schema::SchemaRef;
7use async_trait::async_trait;
8use parking_lot::RwLock;
9
10use crate::config::{ConnectorConfig, ConnectorInfo};
11use crate::connector::{SinkConnector, SourceConnector};
12use crate::error::ConnectorError;
13use crate::reference::ReferenceTableSource;
14use crate::serde::{self, Format, RecordDeserializer, RecordSerializer};
15
16/// Factory function type for creating source connectors.
17///
18/// The optional shared Prometheus registry allows connectors to register
19/// metrics and retain stable registry identity when one is available.
20/// Construction failures, including metrics registration errors, are returned
21/// to the caller.
22pub type SourceFactory = Arc<
23    dyn Fn(Option<&Arc<prometheus::Registry>>) -> Result<Box<dyn SourceConnector>, ConnectorError>
24        + Send
25        + Sync,
26>;
27
28/// Factory function type for creating sink connectors.
29///
30/// The connector config is available during construction so factories can
31/// select a concrete implementation and reject invalid mode-specific options
32/// before `open()` performs external I/O.
33///
34/// The optional shared Prometheus registry allows connectors to register
35/// metrics and retain stable registry identity when one is available.
36pub type SinkFactory = Arc<
37    dyn Fn(
38            &ConnectorConfig,
39            Option<&Arc<prometheus::Registry>>,
40        ) -> Result<Box<dyn SinkConnector>, ConnectorError>
41        + Send
42        + Sync,
43>;
44
45/// Factory for finite reference-table snapshots constrained by the declared schema.
46pub type TableSourceFactory = Arc<
47    dyn Fn(&ConnectorConfig, SchemaRef) -> Result<Box<dyn ReferenceTableSource>, ConnectorError>
48        + Send
49        + Sync,
50>;
51
52/// Factory for constructing a lookup source (async, for on-demand mode).
53#[async_trait]
54pub trait LookupSourceFactory: Send + Sync {
55    /// Build a lookup source instance from the given config.
56    ///
57    /// `declared_schema` is the table's declared Arrow schema, when known.
58    /// Schema-bearing sources may derive their own; schemaless sources use it
59    /// to project records into typed columns.
60    async fn build(
61        &self,
62        config: ConnectorConfig,
63        declared_schema: Option<arrow_schema::SchemaRef>,
64    ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, ConnectorError>;
65}
66
67type LookupSourceRegistration = (ConnectorInfo, Arc<dyn LookupSourceFactory>);
68
69/// Registry of available connector implementations. Connectors register
70/// a factory per type string; the runtime resolves the connector named by
71/// `FROM` in `CREATE SOURCE` or `INTO` in `CREATE SINK`.
72#[derive(Clone)]
73pub struct ConnectorRegistry {
74    sources: Arc<RwLock<HashMap<String, (ConnectorInfo, SourceFactory)>>>,
75    sinks: Arc<RwLock<HashMap<String, (ConnectorInfo, SinkFactory)>>>,
76    table_sources: Arc<RwLock<HashMap<String, (ConnectorInfo, TableSourceFactory)>>>,
77    lookup_sources: Arc<RwLock<HashMap<String, LookupSourceRegistration>>>,
78    /// A registration holds a read guard through insertion; `freeze` takes the write guard.
79    /// This makes the freeze boundary linearizable with concurrent registration attempts.
80    frozen: Arc<RwLock<bool>>,
81}
82
83impl ConnectorRegistry {
84    /// Creates a new empty registry.
85    #[must_use]
86    pub fn new() -> Self {
87        Self {
88            sources: Arc::new(RwLock::new(HashMap::new())),
89            sinks: Arc::new(RwLock::new(HashMap::new())),
90            table_sources: Arc::new(RwLock::new(HashMap::new())),
91            lookup_sources: Arc::new(RwLock::new(HashMap::new())),
92            frozen: Arc::new(RwLock::new(false)),
93        }
94    }
95
96    /// Permanently closes this registry to factory mutation.
97    ///
98    /// The transition is idempotent. Once this call returns, every registration that began
99    /// before it is visible and every later registration is rejected.
100    pub fn freeze(&self) {
101        *self.frozen.write() = true;
102    }
103
104    /// Returns whether factory registration has been permanently disabled.
105    #[must_use]
106    pub fn is_frozen(&self) -> bool {
107        *self.frozen.read()
108    }
109
110    /// Registers a source connector factory.
111    ///
112    /// # Errors
113    ///
114    /// Returns an error if the source name is already registered or the registry is frozen.
115    pub fn register_source(
116        &self,
117        name: impl Into<String>,
118        info: ConnectorInfo,
119        factory: SourceFactory,
120    ) -> Result<(), ConnectorError> {
121        let name = name.into();
122        let frozen = self.frozen.read();
123        if *frozen {
124            return Err(ConnectorError::RegistryFrozen {
125                kind: "source",
126                name,
127            });
128        }
129        let mut sources = self.sources.write();
130        if sources.contains_key(&name) {
131            return Err(ConnectorError::FactoryAlreadyRegistered {
132                kind: "source",
133                name,
134            });
135        }
136        sources.insert(name, (info, factory));
137        drop(frozen);
138        Ok(())
139    }
140
141    /// Registers a sink connector factory.
142    ///
143    /// # Errors
144    ///
145    /// Returns an error if the sink name is already registered or the registry is frozen.
146    pub fn register_sink(
147        &self,
148        name: impl Into<String>,
149        info: ConnectorInfo,
150        factory: SinkFactory,
151    ) -> Result<(), ConnectorError> {
152        let name = name.into();
153        let frozen = self.frozen.read();
154        if *frozen {
155            return Err(ConnectorError::RegistryFrozen { kind: "sink", name });
156        }
157        let mut sinks = self.sinks.write();
158        if sinks.contains_key(&name) {
159            return Err(ConnectorError::FactoryAlreadyRegistered { kind: "sink", name });
160        }
161        sinks.insert(name, (info, factory));
162        drop(frozen);
163        Ok(())
164    }
165
166    /// Run a connector's `discover_schema` against the given properties.
167    ///
168    /// Three outcomes:
169    /// - `Ok(Some(schema))` — discovery succeeded and produced fields.
170    /// - `Ok(None)` — connector type is unknown or does not support discovery.
171    /// - `Err(_)` — discovery was attempted and failed with a specific
172    ///   cause; callers should surface the message verbatim.
173    ///
174    /// # Errors
175    ///
176    /// Returns the source factory's construction error or the underlying
177    /// [`ConnectorError`] from [`SourceConnector::discover_schema`] when
178    /// discovery fails (bad config, unreachable network endpoint, timeout,
179    /// etc.).
180    pub async fn default_source_schema(
181        &self,
182        connector_type: &str,
183        properties: &std::collections::HashMap<String, String>,
184    ) -> Result<Option<SchemaRef>, ConnectorError> {
185        let factory = {
186            let sources = self.sources.read();
187            let Some((_, factory)) = sources.get(connector_type) else {
188                return Ok(None);
189            };
190            factory.clone()
191        };
192
193        let mut instance = factory(None)?;
194        instance.discover_schema(properties).await?;
195        let schema = instance.schema();
196        Ok((!schema.fields().is_empty()).then_some(schema))
197    }
198
199    /// Creates a new source connector instance.
200    ///
201    /// The factory creates a default-configured connector. The caller forwards
202    /// the resolved `FROM` and `FORMAT` configuration in the startup request.
203    ///
204    /// If a `prometheus::Registry` is provided, the connector will register
205    /// its metrics on it so they appear in the scrape output.
206    ///
207    /// # Errors
208    ///
209    /// Returns `ConnectorError::ConfigurationError` if not registered, or the
210    /// source factory's construction error.
211    pub fn create_source(
212        &self,
213        config: &ConnectorConfig,
214        registry: Option<&Arc<prometheus::Registry>>,
215    ) -> Result<Box<dyn SourceConnector>, ConnectorError> {
216        let sources = self.sources.read();
217        let (_, factory) = sources.get(config.connector_type()).ok_or_else(|| {
218            ConnectorError::ConfigurationError(format!(
219                "unknown source connector type: '{}'",
220                config.connector_type()
221            ))
222        })?;
223        factory(registry)
224    }
225
226    /// Derive a registered source connector's semantic recovery identity.
227    ///
228    /// This constructs an unopened source instance and invokes its pure,
229    /// configuration-only identity hook. `None` requests the runtime's
230    /// conservative sanitized-property fallback.
231    ///
232    /// # Errors
233    ///
234    /// Returns an error when the source type is not registered or the connector
235    /// rejects the supplied semantic configuration.
236    pub fn source_recovery_identity_options(
237        &self,
238        config: &ConnectorConfig,
239    ) -> Result<Option<BTreeMap<String, String>>, ConnectorError> {
240        self.create_source(config, None)?
241            .recovery_identity_options(config)
242    }
243
244    /// Creates a new sink connector instance.
245    ///
246    /// The connector type is determined by `config.connector_type()`.
247    ///
248    /// If a `prometheus::Registry` is provided, the connector will register
249    /// its metrics on it so they appear in the scrape output.
250    ///
251    /// # Errors
252    ///
253    /// Returns `ConnectorError::ConfigurationError` if the connector type is
254    /// not registered, or the selected sink configuration is invalid.
255    pub fn create_sink(
256        &self,
257        config: &ConnectorConfig,
258        registry: Option<&Arc<prometheus::Registry>>,
259    ) -> Result<Box<dyn SinkConnector>, ConnectorError> {
260        let sinks = self.sinks.read();
261        let (_, factory) = sinks.get(config.connector_type()).ok_or_else(|| {
262            ConnectorError::ConfigurationError(format!(
263                "unknown sink connector type: '{}'",
264                config.connector_type()
265            ))
266        })?;
267        factory(config, registry)
268    }
269
270    /// Registers a reference table source factory.
271    ///
272    /// # Errors
273    ///
274    /// Returns an error if the table-source name is already registered or the registry is frozen.
275    pub fn register_table_source(
276        &self,
277        name: impl Into<String>,
278        info: ConnectorInfo,
279        factory: TableSourceFactory,
280    ) -> Result<(), ConnectorError> {
281        let name = name.into();
282        let frozen = self.frozen.read();
283        if *frozen {
284            return Err(ConnectorError::RegistryFrozen {
285                kind: "table source",
286                name,
287            });
288        }
289        let mut table_sources = self.table_sources.write();
290        if table_sources.contains_key(&name) {
291            return Err(ConnectorError::FactoryAlreadyRegistered {
292                kind: "table source",
293                name,
294            });
295        }
296        table_sources.insert(name, (info, factory));
297        drop(frozen);
298        Ok(())
299    }
300
301    /// Creates a new reference table source instance.
302    ///
303    /// The connector type is determined by `config.connector_type()`. The declared schema is the
304    /// authoritative field order, Arrow type, and nullability boundary for every returned batch.
305    ///
306    /// # Errors
307    ///
308    /// Returns `ConnectorError::ConfigurationError` if the connector type
309    /// is not registered as a table source.
310    pub fn create_table_source(
311        &self,
312        config: &ConnectorConfig,
313        declared_schema: SchemaRef,
314    ) -> Result<Box<dyn ReferenceTableSource>, ConnectorError> {
315        let table_sources = self.table_sources.read();
316        let (_, factory) = table_sources.get(config.connector_type()).ok_or_else(|| {
317            ConnectorError::ConfigurationError(format!(
318                "connector type '{}' is not registered as a snapshot-capable table source",
319                config.connector_type()
320            ))
321        })?;
322        factory(config, declared_schema)
323    }
324
325    /// Lists all registered table source connector names.
326    #[must_use]
327    pub fn list_table_sources(&self) -> Vec<String> {
328        let mut names: Vec<_> = self.table_sources.read().keys().cloned().collect();
329        names.sort_unstable();
330        names
331    }
332
333    /// Returns whether a snapshot-capable table source is registered.
334    #[must_use]
335    pub fn has_table_source(&self, name: &str) -> bool {
336        self.table_sources.read().contains_key(name)
337    }
338
339    /// Registers a lookup source factory for on-demand/partial cache mode.
340    ///
341    /// # Errors
342    ///
343    /// Returns an error if the lookup name is already registered or the registry is frozen.
344    pub fn register_lookup_source(
345        &self,
346        name: impl Into<String>,
347        info: ConnectorInfo,
348        factory: Arc<dyn LookupSourceFactory>,
349    ) -> Result<(), ConnectorError> {
350        let name = name.into();
351        let frozen = self.frozen.read();
352        if *frozen {
353            return Err(ConnectorError::RegistryFrozen {
354                kind: "lookup source",
355                name,
356            });
357        }
358        let mut lookup_sources = self.lookup_sources.write();
359        if lookup_sources.contains_key(&name) {
360            return Err(ConnectorError::FactoryAlreadyRegistered {
361                kind: "lookup source",
362                name,
363            });
364        }
365        lookup_sources.insert(name, (info, factory));
366        drop(frozen);
367        Ok(())
368    }
369
370    /// Returns whether an on-demand lookup source is registered.
371    #[must_use]
372    pub fn has_lookup_source(&self, name: &str) -> bool {
373        self.lookup_sources.read().contains_key(name)
374    }
375
376    /// Creates a lookup source for on-demand cache-miss fallback.
377    ///
378    /// Returns `None` if no lookup source factory is registered for
379    /// the given connector type.
380    pub async fn create_lookup_source(
381        &self,
382        config: ConnectorConfig,
383        declared_schema: Option<arrow_schema::SchemaRef>,
384    ) -> Option<Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, ConnectorError>>
385    {
386        let factory = {
387            let lookup_sources = self.lookup_sources.read();
388            Arc::clone(&lookup_sources.get(config.connector_type())?.1)
389        };
390        Some(factory.build(config, declared_schema).await)
391    }
392
393    /// Returns information about a registered source connector.
394    #[must_use]
395    pub fn source_info(&self, name: &str) -> Option<ConnectorInfo> {
396        self.sources.read().get(name).map(|(info, _)| info.clone())
397    }
398
399    /// Returns information about a registered sink connector.
400    #[must_use]
401    pub fn sink_info(&self, name: &str) -> Option<ConnectorInfo> {
402        self.sinks.read().get(name).map(|(info, _)| info.clone())
403    }
404
405    /// Lists all registered source connector names.
406    #[must_use]
407    pub fn list_sources(&self) -> Vec<String> {
408        let mut names: Vec<_> = self.sources.read().keys().cloned().collect();
409        names.sort_unstable();
410        names
411    }
412
413    /// Lists all registered sink connector names.
414    #[must_use]
415    pub fn list_sinks(&self) -> Vec<String> {
416        let mut names: Vec<_> = self.sinks.read().keys().cloned().collect();
417        names.sort_unstable();
418        names
419    }
420
421    /// Creates a deserializer for the given format string.
422    ///
423    /// # Errors
424    ///
425    /// Returns `ConnectorError::Serde` if the format is not supported.
426    pub fn create_deserializer(
427        &self,
428        format: &str,
429    ) -> Result<Box<dyn RecordDeserializer>, ConnectorError> {
430        let fmt = Format::parse(format).map_err(ConnectorError::Serde)?;
431        serde::create_deserializer(fmt).map_err(ConnectorError::Serde)
432    }
433
434    /// Creates a serializer for the given format string.
435    ///
436    /// # Errors
437    ///
438    /// Returns `ConnectorError::Serde` if the format is not supported.
439    pub fn create_serializer(
440        &self,
441        format: &str,
442    ) -> Result<Box<dyn RecordSerializer>, ConnectorError> {
443        let fmt = Format::parse(format).map_err(ConnectorError::Serde)?;
444        serde::create_serializer(fmt).map_err(ConnectorError::Serde)
445    }
446}
447
448impl Default for ConnectorRegistry {
449    fn default() -> Self {
450        Self::new()
451    }
452}
453
454impl std::fmt::Debug for ConnectorRegistry {
455    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
456        f.debug_struct("ConnectorRegistry")
457            .field("sources", &self.list_sources())
458            .field("sinks", &self.list_sinks())
459            .field("table_sources", &self.list_table_sources())
460            .finish()
461    }
462}
463
464#[cfg(test)]
465mod tests {
466    use super::*;
467    use crate::reference::MockReferenceTableSource;
468    use crate::testing::*;
469
470    fn mock_info(name: &str, is_source: bool, is_sink: bool) -> ConnectorInfo {
471        ConnectorInfo {
472            name: name.to_string(),
473            display_name: name.to_string(),
474            version: "0.1.0".to_string(),
475            is_source,
476            is_sink,
477            config_keys: vec![],
478        }
479    }
480
481    fn declared_schema() -> SchemaRef {
482        Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
483            "id",
484            arrow_schema::DataType::Int64,
485            false,
486        )]))
487    }
488
489    #[test]
490    fn test_register_and_create_source() {
491        let registry = ConnectorRegistry::new();
492        registry
493            .register_source(
494                "mock",
495                mock_info("mock", true, false),
496                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
497                    Ok(Box::new(MockSourceConnector::new()))
498                }),
499            )
500            .unwrap();
501
502        let config = ConnectorConfig::new("mock");
503        let connector = registry.create_source(&config, None);
504        assert!(connector.is_ok());
505        assert_eq!(
506            registry.source_recovery_identity_options(&config).unwrap(),
507            None
508        );
509    }
510
511    #[test]
512    fn test_register_and_create_sink() {
513        let registry = ConnectorRegistry::new();
514        registry
515            .register_sink(
516                "mock",
517                mock_info("mock", false, true),
518                Arc::new(|_config, _registry| Ok(Box::new(MockSinkConnector::new()))),
519            )
520            .unwrap();
521
522        let config = ConnectorConfig::new("mock");
523        let connector = registry.create_sink(&config, None);
524        assert!(connector.is_ok());
525    }
526
527    #[test]
528    fn factories_receive_shared_registry_identity() {
529        let connectors = ConnectorRegistry::new();
530        let metrics = Arc::new(prometheus::Registry::new());
531
532        let source_metrics = Arc::clone(&metrics);
533        connectors
534            .register_source(
535                "identity-source",
536                mock_info("identity-source", true, false),
537                Arc::new(move |registry: Option<&Arc<prometheus::Registry>>| {
538                    assert!(
539                        registry.is_some_and(|registry| { Arc::ptr_eq(registry, &source_metrics) })
540                    );
541                    Ok(Box::new(MockSourceConnector::new()))
542                }),
543            )
544            .unwrap();
545
546        let sink_metrics = Arc::clone(&metrics);
547        connectors
548            .register_sink(
549                "identity-sink",
550                mock_info("identity-sink", false, true),
551                Arc::new(
552                    move |_config, registry: Option<&Arc<prometheus::Registry>>| {
553                        assert!(registry
554                            .is_some_and(|registry| { Arc::ptr_eq(registry, &sink_metrics) }));
555                        Ok(Box::new(MockSinkConnector::new()))
556                    },
557                ),
558            )
559            .unwrap();
560
561        connectors
562            .create_source(&ConnectorConfig::new("identity-source"), Some(&metrics))
563            .expect("source factory must receive the shared registry Arc");
564        connectors
565            .create_sink(&ConnectorConfig::new("identity-sink"), Some(&metrics))
566            .expect("sink factory must receive the shared registry Arc");
567    }
568
569    struct RejectLookupFactory;
570
571    #[async_trait]
572    impl LookupSourceFactory for RejectLookupFactory {
573        async fn build(
574            &self,
575            _config: ConnectorConfig,
576            _declared_schema: Option<SchemaRef>,
577        ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, ConnectorError>
578        {
579            Err(ConnectorError::ConfigurationError(
580                "test lookup factory has no backing source".into(),
581            ))
582        }
583    }
584
585    #[test]
586    fn sink_factory_receives_config_and_propagates_validation_errors() {
587        let registry = ConnectorRegistry::new();
588        registry
589            .register_sink(
590                "validated",
591                mock_info("validated", false, true),
592                Arc::new(|config, _registry| {
593                    if config.get("enabled") == Some("true") {
594                        Ok(Box::new(MockSinkConnector::new()))
595                    } else {
596                        Err(ConnectorError::ConfigurationError(
597                            "validated sink requires enabled = true".into(),
598                        ))
599                    }
600                }),
601            )
602            .unwrap();
603
604        let mut config = ConnectorConfig::new("validated");
605        let error = match registry.create_sink(&config, None) {
606            Ok(_) => panic!("expected factory validation error"),
607            Err(error) => error,
608        };
609        assert!(error.to_string().contains("enabled = true"));
610
611        config.set("enabled", "true");
612        assert!(registry.create_sink(&config, None).is_ok());
613    }
614
615    #[test]
616    fn test_create_unknown_connector() {
617        let registry = ConnectorRegistry::new();
618        let config = ConnectorConfig::new("nonexistent");
619
620        assert!(registry.create_source(&config, None).is_err());
621        assert!(registry.source_recovery_identity_options(&config).is_err());
622        assert!(registry.create_sink(&config, None).is_err());
623    }
624
625    #[test]
626    fn test_list_connectors() {
627        let registry = ConnectorRegistry::new();
628        registry
629            .register_source(
630                "kafka",
631                mock_info("kafka", true, false),
632                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
633                    Ok(Box::new(MockSourceConnector::new()))
634                }),
635            )
636            .unwrap();
637        registry
638            .register_sink(
639                "delta",
640                mock_info("delta", false, true),
641                Arc::new(|_config, _registry| Ok(Box::new(MockSinkConnector::new()))),
642            )
643            .unwrap();
644
645        let sources = registry.list_sources();
646        assert_eq!(sources.len(), 1);
647        assert!(sources.contains(&"kafka".to_string()));
648
649        let sinks = registry.list_sinks();
650        assert_eq!(sinks.len(), 1);
651        assert!(sinks.contains(&"delta".to_string()));
652    }
653
654    #[test]
655    fn test_connector_info() {
656        let registry = ConnectorRegistry::new();
657        registry
658            .register_source(
659                "kafka",
660                mock_info("kafka", true, false),
661                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
662                    Ok(Box::new(MockSourceConnector::new()))
663                }),
664            )
665            .unwrap();
666
667        let info = registry.source_info("kafka");
668        assert!(info.is_some());
669        assert_eq!(info.unwrap().name, "kafka");
670
671        assert!(registry.source_info("nonexistent").is_none());
672    }
673
674    #[test]
675    fn test_format_registry() {
676        let registry = ConnectorRegistry::new();
677
678        assert!(registry.create_deserializer("json").is_ok());
679        assert!(registry.create_serializer("csv").is_ok());
680        assert!(registry.create_deserializer("unknown").is_err());
681    }
682
683    #[tokio::test]
684    async fn default_source_schema_some_when_discovered() {
685        let registry = ConnectorRegistry::new();
686        registry
687            .register_source(
688                "mock",
689                mock_info("mock", true, false),
690                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
691                    Ok(Box::new(MockSourceConnector::new()))
692                }),
693            )
694            .unwrap();
695        let schema = registry
696            .default_source_schema("mock", &std::collections::HashMap::new())
697            .await
698            .expect("discovery must not fail");
699        assert!(schema.is_some_and(|s| !s.fields().is_empty()));
700    }
701
702    #[tokio::test]
703    async fn default_source_schema_none_for_unknown_connector() {
704        let registry = ConnectorRegistry::new();
705        assert!(registry
706            .default_source_schema("nope", &std::collections::HashMap::new())
707            .await
708            .expect("unknown connector is Ok(None), not Err")
709            .is_none());
710    }
711
712    #[tokio::test]
713    async fn source_factory_errors_propagate_from_creation_and_discovery() {
714        let registry = ConnectorRegistry::new();
715        registry
716            .register_source(
717                "failing",
718                mock_info("failing", true, false),
719                Arc::new(
720                    |_: Option<&Arc<prometheus::Registry>>| -> Result<
721                        Box<dyn SourceConnector>,
722                        ConnectorError,
723                    > {
724                        Err(ConnectorError::Internal("source construction failed".into()))
725                    },
726                ),
727            )
728            .unwrap();
729
730        let config = ConnectorConfig::new("failing");
731        let Err(create_error) = registry.create_source(&config, None) else {
732            panic!("source construction must fail");
733        };
734        assert!(create_error
735            .to_string()
736            .contains("source construction failed"));
737
738        let discovery_error = registry
739            .default_source_schema("failing", &std::collections::HashMap::new())
740            .await
741            .expect_err("schema discovery must propagate source construction failure");
742        assert!(discovery_error
743            .to_string()
744            .contains("source construction failed"));
745    }
746
747    // ── Table source factory tests ──
748
749    #[test]
750    fn test_register_and_create_table_source() {
751        use crate::reference::MockReferenceTableSource;
752
753        let registry = ConnectorRegistry::new();
754        let observed_schema = Arc::new(parking_lot::Mutex::new(None));
755        let factory_schema = Arc::clone(&observed_schema);
756        registry
757            .register_table_source(
758                "mock",
759                mock_info("mock", true, false),
760                Arc::new(move |_config, declared_schema| {
761                    *factory_schema.lock() = Some(declared_schema);
762                    Ok(Box::new(MockReferenceTableSource::empty()))
763                }),
764            )
765            .unwrap();
766
767        let config = ConnectorConfig::new("mock");
768        let declared_schema = declared_schema();
769        let source = registry.create_table_source(&config, Arc::clone(&declared_schema));
770        assert!(source.is_ok());
771        assert_eq!(observed_schema.lock().as_ref(), Some(&declared_schema));
772        assert!(registry.has_table_source("mock"));
773        assert!(!registry.has_table_source("missing"));
774    }
775
776    #[test]
777    fn test_create_unknown_table_source() {
778        let registry = ConnectorRegistry::new();
779        let config = ConnectorConfig::new("nonexistent");
780        let result = registry.create_table_source(&config, declared_schema());
781        match result {
782            Err(e) => assert!(
783                e.to_string().contains("snapshot-capable table source"),
784                "got: {e}"
785            ),
786            Ok(_) => panic!("Expected error for unknown table source"),
787        }
788    }
789
790    #[test]
791    fn test_list_table_sources() {
792        let registry = ConnectorRegistry::new();
793        assert!(registry.list_table_sources().is_empty());
794
795        registry
796            .register_table_source(
797                "mock-table",
798                mock_info("mock-table", true, false),
799                Arc::new(|_config, _declared_schema| {
800                    Ok(Box::new(MockReferenceTableSource::empty()))
801                }),
802            )
803            .unwrap();
804
805        let names = registry.list_table_sources();
806        assert_eq!(names.len(), 1);
807        assert!(names.contains(&"mock-table".to_string()));
808    }
809
810    #[test]
811    fn duplicate_registration_is_rejected_in_every_category() {
812        let registry = ConnectorRegistry::new();
813        let source = || {
814            Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
815                Ok(Box::new(MockSourceConnector::new()) as Box<dyn SourceConnector>)
816            }) as SourceFactory
817        };
818        let sink = || {
819            Arc::new(
820                |_config: &ConnectorConfig, _registry: Option<&Arc<prometheus::Registry>>| {
821                    Ok(Box::new(MockSinkConnector::new()) as Box<dyn SinkConnector>)
822                },
823            ) as SinkFactory
824        };
825        let table = || {
826            Arc::new(|_config: &ConnectorConfig, _declared_schema: SchemaRef| {
827                Ok(Box::new(MockReferenceTableSource::empty()) as Box<dyn ReferenceTableSource>)
828            }) as TableSourceFactory
829        };
830
831        registry
832            .register_source("same", mock_info("same", true, false), source())
833            .unwrap();
834        assert!(matches!(
835            registry.register_source("same", mock_info("same", true, false), source()),
836            Err(ConnectorError::FactoryAlreadyRegistered { kind: "source", .. })
837        ));
838
839        registry
840            .register_sink("same", mock_info("same", false, true), sink())
841            .unwrap();
842        assert!(matches!(
843            registry.register_sink("same", mock_info("same", false, true), sink()),
844            Err(ConnectorError::FactoryAlreadyRegistered { kind: "sink", .. })
845        ));
846
847        registry
848            .register_table_source("same", mock_info("same", true, false), table())
849            .unwrap();
850        assert!(matches!(
851            registry.register_table_source("same", mock_info("same", true, false), table()),
852            Err(ConnectorError::FactoryAlreadyRegistered {
853                kind: "table source",
854                ..
855            })
856        ));
857
858        registry
859            .register_lookup_source(
860                "same",
861                mock_info("same", true, false),
862                Arc::new(RejectLookupFactory),
863            )
864            .unwrap();
865        assert!(matches!(
866            registry.register_lookup_source(
867                "same",
868                mock_info("same", true, false),
869                Arc::new(RejectLookupFactory)
870            ),
871            Err(ConnectorError::FactoryAlreadyRegistered {
872                kind: "lookup source",
873                ..
874            })
875        ));
876    }
877
878    #[test]
879    fn freeze_rejects_every_registration_category() {
880        let registry = ConnectorRegistry::new();
881        registry.freeze();
882        registry.freeze();
883        assert!(registry.is_frozen());
884
885        assert!(matches!(
886            registry.register_source(
887                "late-source",
888                mock_info("late-source", true, false),
889                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
890                    Ok(Box::new(MockSourceConnector::new()))
891                })
892            ),
893            Err(ConnectorError::RegistryFrozen { kind: "source", .. })
894        ));
895        assert!(matches!(
896            registry.register_sink(
897                "late-sink",
898                mock_info("late-sink", false, true),
899                Arc::new(|_config, _registry| Ok(Box::new(MockSinkConnector::new())))
900            ),
901            Err(ConnectorError::RegistryFrozen { kind: "sink", .. })
902        ));
903        assert!(matches!(
904            registry.register_table_source(
905                "late-table",
906                mock_info("late-table", true, false),
907                Arc::new(|_config, _declared_schema| {
908                    Ok(Box::new(MockReferenceTableSource::empty()))
909                })
910            ),
911            Err(ConnectorError::RegistryFrozen {
912                kind: "table source",
913                ..
914            })
915        ));
916        assert!(matches!(
917            registry.register_lookup_source(
918                "late-lookup",
919                mock_info("late-lookup", true, false),
920                Arc::new(RejectLookupFactory)
921            ),
922            Err(ConnectorError::RegistryFrozen {
923                kind: "lookup source",
924                ..
925            })
926        ));
927    }
928}
laminar_connectors/registry.rs

laminar_connectors/
registry.rs
