danube-connect-core 0.5.0

Core SDK for building Danube connectors
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

//! # Danube Connect Core
//!
//! Core SDK for building Danube connectors.
//!
//! This library provides the foundational framework for creating connectors that integrate
//! external systems with Danube Messaging. It handles all the complexity of communicating
//! with Danube brokers, managing lifecycles, retries, and observability, allowing connector
//! developers to focus solely on the integration logic.
//!
//! ## Overview
//!
//! Connectors are standalone processes that either:
//! - **Sink**: Consume messages from Danube and write to an external system
//! - **Source**: Read from an external system and publish to Danube
//!
//! ## Quick Start
//!
//! ```rust,no_run
//! use danube_connect_core::{SinkConnector, SinkRecord, ConnectorConfig, ConnectorResult, ConsumerConfig, SubscriptionType};
//! use async_trait::async_trait;
//!
//! pub struct MyConnector;
//!
//! #[async_trait]
//! impl SinkConnector for MyConnector {
//!     async fn initialize(&mut self, config: ConnectorConfig) -> ConnectorResult<()> {
//!         // Setup your connector
//!         Ok(())
//!     }
//!     
//!     async fn consumer_configs(&self) -> ConnectorResult<Vec<ConsumerConfig>> {
//!         Ok(vec![ConsumerConfig {
//!             topic: "/default/my-topic".to_string(),
//!             consumer_name: "my-consumer".to_string(),
//!             subscription: "my-sub".to_string(),
//!             subscription_type: SubscriptionType::Exclusive,
//!             expected_schema_subject: None,
//!         }])
//!     }
//!     
//!     async fn process_batch(&mut self, records: Vec<SinkRecord>) -> ConnectorResult<()> {
//!         // Process message batch
//!         for record in records {
//!             println!("Got message: {:?}", record.payload());
//!         }
//!         Ok(())
//!     }
//! }
//! ```
//!
//! ## Features
//!
//! - **Automatic Lifecycle Management**: The runtime handles initialization, message loops, and shutdown
//! - **Built-in Retry Logic**: Configurable exponential backoff for transient failures
//! - **Message Transformation**: Helpers for JSON, binary, and schema-based transformations
//! - **Observability**: Automatic metrics, structured logging, and health checks
//! - **Configuration**: Standard environment variable and file-based configuration

mod config;
mod error;
mod message;
mod metrics;
mod retry;
mod route;
mod runtime;
mod schema;
mod traits;
pub mod utils;

// Re-export public API
pub use config::{
    ConfigEnvOverrides, ConfigValidate, ConnectorConfig, ConnectorConfigLoader, ConsumerConfig,
    ProcessingSettings, ProducerConfig, RetrySettings, SchemaConfig, SchemaMapping,
    SubscriptionType, VersionStrategy,
};
pub use error::{ConnectorError, ConnectorResult};
pub use message::{DanubeMetadata, RecordContext, RoutingContext, SinkRecord, SourceRecord};
pub use metrics::ConnectorMetrics;
pub use route::{
    RouteDispatchPolicy, RouteSchemaPolicy, RouteSubscriptionPolicy, SinkRoute, SourceRoute,
};
pub use runtime::{SinkRuntime, SourceRuntime};
pub use schema::SchemaType;
pub use traits::{
    Offset, SinkConnector, SourceConnector, SourceConnectorMode, SourceEnvelope, SourceSender,
};
pub use utils::{Batcher, HealthChecker, HealthStatus};

// Re-export commonly used types from danube-client
pub use danube_client::{SchemaInfo, SubType};

// Version info
/// Crate version for the currently built `danube-connect-core` package.
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
/// Crate name for the currently built `danube-connect-core` package.
pub const NAME: &str = env!("CARGO_PKG_NAME");