Allow creating replication slots using multiple publications.

examples/basic.rs

@@ -33,7 +33,7 @@ pub async fn main() -> anyhow::Result<()> {
         database,
         tls: TlsConfig::disabled(),
         slot,
-        publication,
+        publication: publication.into(),
         start_lsn,
         stop_at_lsn: None,
 

examples/bounded_replay.rs

@@ -45,7 +45,7 @@ pub async fn main() -> anyhow::Result<()> {
         database,
         tls: TlsConfig::disabled(),
         slot,
-        publication,
+        publication: publication.into(),
         start_lsn,
         stop_at_lsn: Some(stop_at_lsn),
 

examples/checkpointed.rs

@@ -36,7 +36,7 @@ async fn main() -> anyhow::Result<()> {
         database,
         tls: TlsConfig::disabled(),
         slot,
-        publication,
+        publication: publication.into(),
         start_lsn,
         stop_at_lsn: None,
         status_interval: std::time::Duration::from_secs(1),

examples/with_mtls.rs

@@ -67,7 +67,7 @@ pub async fn main() -> anyhow::Result<()> {
         },
 
         slot,
-        publication,
+        publication: publication.into(),
         start_lsn,
         stop_at_lsn: None,
 

examples/with_tls.rs

@@ -51,7 +51,7 @@ pub async fn main() -> anyhow::Result<()> {
         },
 
         slot,
-        publication,
+        publication: publication.into(),
         start_lsn,
         stop_at_lsn: None,
 

src/client/worker.rs

@@ -180,12 +180,10 @@ impl WorkerState {
         &self,
         stream: &mut S,
     ) -> Result<()> {
-        // Escape single quotes in publication name
-        let publication = self.cfg.publication.replace('\'', "''");
         let sql = format!(
             "START_REPLICATION SLOT {} LOGICAL {} \
             (proto_version '1', publication_names '{}', messages 'true')",
-            self.cfg.slot, self.cfg.start_lsn, publication,
+            self.cfg.slot, self.cfg.start_lsn, self.cfg.publication,
         );
         write_query(stream, &sql).await?;
 

src/config.rs

@@ -214,6 +214,34 @@ impl TlsConfig {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Publication(Vec<String>);
+
+impl From<&str> for Publication {
+    fn from(value: &str) -> Self {
+        Self(vec![value.into()])
+    }
+}
+
+impl From<String> for Publication {
+    fn from(value: String) -> Self {
+        Self(vec![value])
+    }
+}
+
+impl<A: Into<String>> FromIterator<A> for Publication {
+    fn from_iter<T: IntoIterator<Item = A>>(iter: T) -> Self {
+        Self(iter.into_iter().map(|item| item.into()).collect())
+    }
+}
+
+impl std::fmt::Display for Publication {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // Escape single quotes in publication name
+        write!(f, "{}", self.0.join(",").replace('\'', "''"))
+    }
+}
+
 /// Configuration for PostgreSQL logical replication connections.
 ///
 /// # Example
@@ -267,7 +295,7 @@ pub struct ReplicationConfig {
     /// Name of the publication to subscribe to.
     ///
     /// The publication must exist and include the tables you want to replicate.
-    pub publication: String,
+    pub publication: Publication,
 
     /// LSN position to start replication from.
     ///
@@ -359,7 +387,7 @@ impl ReplicationConfig {
         password: impl Into<String>,
         database: impl Into<String>,
         slot: impl Into<String>,
-        publication: impl Into<String>,
+        publication: impl Into<Publication>,
     ) -> Self {
         Self {
             host: host.into(),
@@ -425,7 +453,7 @@ impl ReplicationConfig {
         password: impl Into<String>,
         database: impl Into<String>,
         slot: impl Into<String>,
-        publication: impl Into<String>,
+        publication: impl Into<Publication>,
     ) -> Self {
         Self {
             host: socket_dir.into(),