pg2any_lib/
lib.rs

1//! # PostgreSQL CDC Library
2//!
3//! A comprehensive Change Data Capture (CDC) library for PostgreSQL using logical replication.
4//! This library allows you to stream database changes in real-time from PostgreSQL to other databases
5//! such as SQL Server, MySQL, and more.
6//!
7
8//! ## Features
9//!
10//! - PostgreSQL logical replication support
11//! - Real-time change streaming (INSERT, UPDATE, DELETE, TRUNCATE)
12//! - Multiple destination database support (SQL Server, MySQL)
13//! - Async/await support with Tokio
14//! - Comprehensive error handling
15//! - Thread-safe operations
16//! - Built-in backpressure handling
17//!
18//! ## Quick Start
19//!
20//! ```rust,ignore
21//! use pg2any_lib::{load_config_from_env, run_cdc_app};
22//! use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
23//!
24//! #[tokio::main]
25//!     async fn main() -> Result<(), Box<dyn std::error::Error>> {
26//!     // Initialize comprehensive logging
27//!     init_logging();
28//!     tracing::info!("Starting PostgreSQL CDC Application");
29//!     // Load configuration from environment variables
30//!     let config = load_config_from_env()?;
31//!     // Run the CDC application with graceful shutdown handling
32//!     run_cdc_app(config, None).await?;
33//!     tracing::info!("CDC application stopped");
34//!     Ok(())
35//! }
36//!
37//! pub fn init_logging() {
38//!     // Create a sophisticated logging setup
39//!     let env_filter = EnvFilter::try_from_default_env()
40//!         .unwrap_or_else(|_| EnvFilter::new("pg2any=debug,tokio_postgres=info,sqlx=info"));
41//!
42//!     let fmt_layer = fmt::layer()
43//!         .with_target(true)
44//!         .with_thread_ids(true)
45//!         .with_level(true)
46//!         .with_ansi(true)
47//!         .compact();
48//!
49//!     tracing_subscriber::registry()
50//!         .with(env_filter)
51//!         .with(fmt_layer)
52//!         .init();
53//!
54//!     tracing::info!("Logging initialized with level filtering");
55//! }
56//! ```
57
58// Core modules
59pub mod app;
60pub mod config;
61pub mod env;
62pub mod error;
63
64// Destination handlers
65pub mod types;
66
67pub mod lsn_tracker;
68
69// Slot-first LSN recovery (queries pg_replication_slots, reconciles with disk)
70pub(crate) mod slot;
71
72// High-level client interface
73pub mod client;
74mod consumer;
75mod producer;
76
77// Transaction file persistence
78pub mod transaction_manager;
79
80// Pure SQL rendering (extracted from transaction_manager)
81pub mod sql_renderer;
82
83// Storage abstraction for transaction files
84pub mod storage;
85
86// Monitoring and metrics
87pub mod monitoring;
88
89// Public API exports
90pub use app::{run_cdc_app, CdcApp, CdcAppConfig};
91pub use client::CdcClient;
92pub use config::{Config, ConfigBuilder};
93pub use consumer::drain_and_shutdown;
94pub use env::load_config_from_env;
95pub use error::CdcError;
96pub use lsn_tracker::{create_lsn_tracker_with_load, LsnTracker};
97pub type CdcResult<T> = Result<T, CdcError>;
98
99pub mod destinations;
100
101pub use pg_walstream::{
102    // Type aliases and utilities
103    format_lsn,
104    // Protocol types
105    message_types,
106    parse_lsn,
107    postgres_timestamp_to_chrono,
108    system_time_to_postgres_timestamp,
109    // Buffer types
110    BufferReader,
111    BufferWriter,
112    // Cancellation token
113    CancellationToken,
114    ColumnData,
115    ColumnInfo,
116    KeepaliveMessage,
117    LogicalReplicationMessage,
118    LogicalReplicationParser,
119    LogicalReplicationStream,
120    // PostgreSQL-specific types
121    Lsn,
122    MessageType,
123    Oid,
124    RelationInfo,
125    ReplicaIdentity,
126    ReplicationState,
127    ReplicationStreamConfig,
128    StreamingReplicationMessage,
129    TimestampTz,
130    TupleData,
131    XLogRecPtr,
132    Xid,
133    INVALID_XLOG_REC_PTR,
134    PG_EPOCH_OFFSET_SECS,
135};
136
137// Re-export SharedLsnFeedback from lsn_tracker (pg2any-lib's version with log_status method)
138pub use lsn_tracker::SharedLsnFeedback;
139
140// Re-export implementations
141#[cfg(feature = "mysql")]
142pub use crate::destinations::MySQLDestination;
143
144#[cfg(feature = "sqlserver")]
145pub use crate::destinations::SqlServerDestination;
146
147#[cfg(feature = "sqlite")]
148pub use crate::destinations::SQLiteDestination;
149
150#[cfg(feature = "kafka")]
151pub use crate::destinations::KafkaDestination;
152
153pub use crate::destinations::{DestinationFactoryFn, DestinationHandler, PreCommitHook};
154pub use crate::types::{DestinationType, Transaction};
155
156// SQL dialect customization for external destinations.
157//
158// Custom destinations registered via `Config::custom_destination` can
159// override `DestinationHandler::dialect()` to return one of the built-in
160// dialects (e.g. `MySqlDialect`), the generic `AnsiDialect`, or a fully
161// custom `SqlDialect` impl. See `destinations::dialect` for the integration
162// guide and `examples/src/bin/custom_destination.rs` for a worked example.
163pub use crate::destinations::dialect::SqlDialect;
164pub use crate::destinations::dialects::{
165    AnsiDialect, KafkaDialect, MySqlDialect, SqlServerDialect, SqliteDialect,
166};
167pub use crate::sql_renderer::{RenderContext, RenderedStatement};
168
169// Conditionally export metrics server functionality
170#[cfg(feature = "metrics")]
171pub use crate::monitoring::{
172    create_metrics_server, create_metrics_server_with_config, init_real_metrics, MetricsServer,
173    MetricsServerConfig,
174};
175
176// Always export metrics abstraction layer
177pub use crate::monitoring::{
178    gather_metrics, init_metrics, MetricsCollector, MetricsCollectorTrait, ProcessingTimer,
179    ProcessingTimerTrait,
180};
pg2any_lib/lib.rs

pg2any_lib/
lib.rs
