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 looks up by the `connector`
71/// property in `CREATE SOURCE/SINK` DDL.
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 must
202    /// subsequently call `open(config)` to forward WITH clause properties.
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!(
554                            registry.is_some_and(|registry| Arc::ptr_eq(registry, &sink_metrics))
555                        );
556                        Ok(Box::new(MockSinkConnector::new()))
557                    },
558                ),
559            )
560            .unwrap();
561
562        connectors
563            .create_source(&ConnectorConfig::new("identity-source"), Some(&metrics))
564            .expect("source factory must receive the shared registry Arc");
565        connectors
566            .create_sink(&ConnectorConfig::new("identity-sink"), Some(&metrics))
567            .expect("sink factory must receive the shared registry Arc");
568    }
569
570    struct RejectLookupFactory;
571
572    #[async_trait]
573    impl LookupSourceFactory for RejectLookupFactory {
574        async fn build(
575            &self,
576            _config: ConnectorConfig,
577            _declared_schema: Option<SchemaRef>,
578        ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, ConnectorError>
579        {
580            Err(ConnectorError::ConfigurationError(
581                "test lookup factory has no backing source".into(),
582            ))
583        }
584    }
585
586    #[test]
587    fn sink_factory_receives_config_and_propagates_validation_errors() {
588        let registry = ConnectorRegistry::new();
589        registry
590            .register_sink(
591                "validated",
592                mock_info("validated", false, true),
593                Arc::new(|config, _registry| {
594                    if config.get("enabled") == Some("true") {
595                        Ok(Box::new(MockSinkConnector::new()))
596                    } else {
597                        Err(ConnectorError::ConfigurationError(
598                            "validated sink requires enabled = true".into(),
599                        ))
600                    }
601                }),
602            )
603            .unwrap();
604
605        let mut config = ConnectorConfig::new("validated");
606        let error = match registry.create_sink(&config, None) {
607            Ok(_) => panic!("expected factory validation error"),
608            Err(error) => error,
609        };
610        assert!(error.to_string().contains("enabled = true"));
611
612        config.set("enabled", "true");
613        assert!(registry.create_sink(&config, None).is_ok());
614    }
615
616    #[test]
617    fn test_create_unknown_connector() {
618        let registry = ConnectorRegistry::new();
619        let config = ConnectorConfig::new("nonexistent");
620
621        assert!(registry.create_source(&config, None).is_err());
622        assert!(registry.source_recovery_identity_options(&config).is_err());
623        assert!(registry.create_sink(&config, None).is_err());
624    }
625
626    #[test]
627    fn test_list_connectors() {
628        let registry = ConnectorRegistry::new();
629        registry
630            .register_source(
631                "kafka",
632                mock_info("kafka", true, false),
633                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
634                    Ok(Box::new(MockSourceConnector::new()))
635                }),
636            )
637            .unwrap();
638        registry
639            .register_sink(
640                "delta",
641                mock_info("delta", false, true),
642                Arc::new(|_config, _registry| Ok(Box::new(MockSinkConnector::new()))),
643            )
644            .unwrap();
645
646        let sources = registry.list_sources();
647        assert_eq!(sources.len(), 1);
648        assert!(sources.contains(&"kafka".to_string()));
649
650        let sinks = registry.list_sinks();
651        assert_eq!(sinks.len(), 1);
652        assert!(sinks.contains(&"delta".to_string()));
653    }
654
655    #[test]
656    fn test_connector_info() {
657        let registry = ConnectorRegistry::new();
658        registry
659            .register_source(
660                "kafka",
661                mock_info("kafka", true, false),
662                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
663                    Ok(Box::new(MockSourceConnector::new()))
664                }),
665            )
666            .unwrap();
667
668        let info = registry.source_info("kafka");
669        assert!(info.is_some());
670        assert_eq!(info.unwrap().name, "kafka");
671
672        assert!(registry.source_info("nonexistent").is_none());
673    }
674
675    #[test]
676    fn test_format_registry() {
677        let registry = ConnectorRegistry::new();
678
679        assert!(registry.create_deserializer("json").is_ok());
680        assert!(registry.create_serializer("csv").is_ok());
681        assert!(registry.create_deserializer("unknown").is_err());
682    }
683
684    #[tokio::test]
685    async fn default_source_schema_some_when_discovered() {
686        let registry = ConnectorRegistry::new();
687        registry
688            .register_source(
689                "mock",
690                mock_info("mock", true, false),
691                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
692                    Ok(Box::new(MockSourceConnector::new()))
693                }),
694            )
695            .unwrap();
696        let schema = registry
697            .default_source_schema("mock", &std::collections::HashMap::new())
698            .await
699            .expect("discovery must not fail");
700        assert!(schema.is_some_and(|s| !s.fields().is_empty()));
701    }
702
703    #[tokio::test]
704    async fn default_source_schema_none_for_unknown_connector() {
705        let registry = ConnectorRegistry::new();
706        assert!(registry
707            .default_source_schema("nope", &std::collections::HashMap::new())
708            .await
709            .expect("unknown connector is Ok(None), not Err")
710            .is_none());
711    }
712
713    #[tokio::test]
714    async fn source_factory_errors_propagate_from_creation_and_discovery() {
715        let registry = ConnectorRegistry::new();
716        registry
717            .register_source(
718                "failing",
719                mock_info("failing", true, false),
720                Arc::new(
721                    |_: Option<&Arc<prometheus::Registry>>| -> Result<
722                        Box<dyn SourceConnector>,
723                        ConnectorError,
724                    > {
725                        Err(ConnectorError::Internal(
726                            "source construction failed".into(),
727                        ))
728                    },
729                ),
730            )
731            .unwrap();
732
733        let config = ConnectorConfig::new("failing");
734        let Err(create_error) = registry.create_source(&config, None) else {
735            panic!("source construction must fail");
736        };
737        assert!(create_error
738            .to_string()
739            .contains("source construction failed"));
740
741        let discovery_error = registry
742            .default_source_schema("failing", &std::collections::HashMap::new())
743            .await
744            .expect_err("schema discovery must propagate source construction failure");
745        assert!(discovery_error
746            .to_string()
747            .contains("source construction failed"));
748    }
749
750    // ── Table source factory tests ──
751
752    #[test]
753    fn test_register_and_create_table_source() {
754        use crate::reference::MockReferenceTableSource;
755
756        let registry = ConnectorRegistry::new();
757        let observed_schema = Arc::new(parking_lot::Mutex::new(None));
758        let factory_schema = Arc::clone(&observed_schema);
759        registry
760            .register_table_source(
761                "mock",
762                mock_info("mock", true, false),
763                Arc::new(move |_config, declared_schema| {
764                    *factory_schema.lock() = Some(declared_schema);
765                    Ok(Box::new(MockReferenceTableSource::empty()))
766                }),
767            )
768            .unwrap();
769
770        let config = ConnectorConfig::new("mock");
771        let declared_schema = declared_schema();
772        let source = registry.create_table_source(&config, Arc::clone(&declared_schema));
773        assert!(source.is_ok());
774        assert_eq!(observed_schema.lock().as_ref(), Some(&declared_schema));
775        assert!(registry.has_table_source("mock"));
776        assert!(!registry.has_table_source("missing"));
777    }
778
779    #[test]
780    fn test_create_unknown_table_source() {
781        let registry = ConnectorRegistry::new();
782        let config = ConnectorConfig::new("nonexistent");
783        let result = registry.create_table_source(&config, declared_schema());
784        match result {
785            Err(e) => assert!(
786                e.to_string().contains("snapshot-capable table source"),
787                "got: {e}"
788            ),
789            Ok(_) => panic!("Expected error for unknown table source"),
790        }
791    }
792
793    #[test]
794    fn test_list_table_sources() {
795        let registry = ConnectorRegistry::new();
796        assert!(registry.list_table_sources().is_empty());
797
798        registry
799            .register_table_source(
800                "mock-table",
801                mock_info("mock-table", true, false),
802                Arc::new(|_config, _declared_schema| {
803                    Ok(Box::new(MockReferenceTableSource::empty()))
804                }),
805            )
806            .unwrap();
807
808        let names = registry.list_table_sources();
809        assert_eq!(names.len(), 1);
810        assert!(names.contains(&"mock-table".to_string()));
811    }
812
813    #[test]
814    fn duplicate_registration_is_rejected_in_every_category() {
815        let registry = ConnectorRegistry::new();
816        let source = || {
817            Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
818                Ok(Box::new(MockSourceConnector::new()) as Box<dyn SourceConnector>)
819            }) as SourceFactory
820        };
821        let sink = || {
822            Arc::new(
823                |_config: &ConnectorConfig, _registry: Option<&Arc<prometheus::Registry>>| {
824                    Ok(Box::new(MockSinkConnector::new()) as Box<dyn SinkConnector>)
825                },
826            ) as SinkFactory
827        };
828        let table = || {
829            Arc::new(|_config: &ConnectorConfig, _declared_schema: SchemaRef| {
830                Ok(Box::new(MockReferenceTableSource::empty()) as Box<dyn ReferenceTableSource>)
831            }) as TableSourceFactory
832        };
833
834        registry
835            .register_source("same", mock_info("same", true, false), source())
836            .unwrap();
837        assert!(matches!(
838            registry.register_source("same", mock_info("same", true, false), source()),
839            Err(ConnectorError::FactoryAlreadyRegistered { kind: "source", .. })
840        ));
841
842        registry
843            .register_sink("same", mock_info("same", false, true), sink())
844            .unwrap();
845        assert!(matches!(
846            registry.register_sink("same", mock_info("same", false, true), sink()),
847            Err(ConnectorError::FactoryAlreadyRegistered { kind: "sink", .. })
848        ));
849
850        registry
851            .register_table_source("same", mock_info("same", true, false), table())
852            .unwrap();
853        assert!(matches!(
854            registry.register_table_source("same", mock_info("same", true, false), table()),
855            Err(ConnectorError::FactoryAlreadyRegistered {
856                kind: "table source",
857                ..
858            })
859        ));
860
861        registry
862            .register_lookup_source(
863                "same",
864                mock_info("same", true, false),
865                Arc::new(RejectLookupFactory),
866            )
867            .unwrap();
868        assert!(matches!(
869            registry.register_lookup_source(
870                "same",
871                mock_info("same", true, false),
872                Arc::new(RejectLookupFactory)
873            ),
874            Err(ConnectorError::FactoryAlreadyRegistered {
875                kind: "lookup source",
876                ..
877            })
878        ));
879    }
880
881    #[test]
882    fn freeze_rejects_every_registration_category() {
883        let registry = ConnectorRegistry::new();
884        registry.freeze();
885        registry.freeze();
886        assert!(registry.is_frozen());
887
888        assert!(matches!(
889            registry.register_source(
890                "late-source",
891                mock_info("late-source", true, false),
892                Arc::new(|_: Option<&Arc<prometheus::Registry>>| {
893                    Ok(Box::new(MockSourceConnector::new()))
894                })
895            ),
896            Err(ConnectorError::RegistryFrozen { kind: "source", .. })
897        ));
898        assert!(matches!(
899            registry.register_sink(
900                "late-sink",
901                mock_info("late-sink", false, true),
902                Arc::new(|_config, _registry| Ok(Box::new(MockSinkConnector::new())))
903            ),
904            Err(ConnectorError::RegistryFrozen { kind: "sink", .. })
905        ));
906        assert!(matches!(
907            registry.register_table_source(
908                "late-table",
909                mock_info("late-table", true, false),
910                Arc::new(|_config, _declared_schema| {
911                    Ok(Box::new(MockReferenceTableSource::empty()))
912                })
913            ),
914            Err(ConnectorError::RegistryFrozen {
915                kind: "table source",
916                ..
917            })
918        ));
919        assert!(matches!(
920            registry.register_lookup_source(
921                "late-lookup",
922                mock_info("late-lookup", true, false),
923                Arc::new(RejectLookupFactory)
924            ),
925            Err(ConnectorError::RegistryFrozen {
926                kind: "lookup source",
927                ..
928            })
929        ));
930    }
931}
laminar_connectors/registry.rs

laminar_connectors/
registry.rs
