add replication test

Author: jeff-davis

docker-compose.yml

@@ -3,4 +3,6 @@ services:
   postgres:
     image: "sfackler/rust-postgres-test:6"
     ports:
-    - 5433:5433
+      - 5433:5433
+    volumes:
+      - "./docker/sql_setup.sh:/docker-entrypoint-initdb.d/sql_setup.sh"

docker/sql_setup.sh

@@ -64,6 +64,7 @@ port = 5433
 ssl = on
 ssl_cert_file = 'server.crt'
 ssl_key_file = 'server.key'
+wal_level = logical
 EOCONF
 
 cat > "$PGDATA/pg_hba.conf" <<-EOCONF
@@ -82,6 +83,7 @@ host    all             ssl_user        ::0/0                reject
 
 # IPv4 local connections:
 host    all             postgres        0.0.0.0/0            trust
+host    replication     postgres        0.0.0.0/0            trust
 # IPv6 local connections:
 host    all             postgres        ::0/0                trust
 # Unix socket connections:

tokio-postgres/src/replication_client.rs

@@ -21,7 +21,7 @@ pub struct IdentifySystem {
     systemid: String,
     timeline: u32,
     xlogpos: Lsn,
-    dbname: String,
+    dbname: Option<String>,
 }
 
 impl IdentifySystem {
@@ -37,8 +37,8 @@ impl IdentifySystem {
         self.xlogpos
     }
 
-    pub fn dbname(&self) -> &str {
-        &self.dbname
+    pub fn dbname(&self) -> Option<&str> {
+        self.dbname.as_deref()
     }
 }
 
@@ -89,16 +89,20 @@ impl ReplicationClient {
             )));
         };
 
-        let str_values = ranges
+        let values: Vec<Option<&str>> = ranges
             .iter()
-            .map(|r| from_utf8(&datarow.buffer()[r.to_owned().unwrap()]).unwrap())
+            .map(|range| {
+                range
+                    .to_owned()
+                    .map(|r| from_utf8(&datarow.buffer()[r]).unwrap())
+            })
             .collect::<Vec<_>>();
 
         Ok(IdentifySystem {
-            systemid: String::from(str_values[0]),
-            timeline: str_values[1].parse::<u32>().unwrap(),
-            xlogpos: Lsn::from(str_values[2]),
-            dbname: String::from(str_values[3]),
+            systemid: String::from(values[0].unwrap()),
+            timeline: values[1].unwrap().parse::<u32>().unwrap(),
+            xlogpos: Lsn::from(values[2].unwrap()),
+            dbname: values[3].map(String::from),
         })
     }
 

tokio-postgres/tests/test/main.rs

@@ -18,6 +18,7 @@ use tokio_postgres::{
 
 mod binary_copy;
 mod parse;
+mod replication;
 #[cfg(feature = "runtime")]
 mod runtime;
 mod types;

tokio-postgres/tests/test/replication.rs

@@ -0,0 +1,66 @@
+use postgres_protocol::message::backend::ReplicationMessage;
+use tokio::stream::StreamExt;
+use tokio_postgres::{connect, connect_replication, NoTls, ReplicationMode};
+
+// test for:
+//   - identify_system
+//   - show
+//   X timeline history
+//   - physical replication
+#[tokio::test]
+async fn physical_replication() {
+    let conninfo = "host=127.0.0.1 port=5433 user=postgres";
+
+    // form SQL connection
+    let (sclient, sconnection) = connect(conninfo, NoTls).await.unwrap();
+    tokio::spawn(async move {
+        if let Err(e) = sconnection.await {
+            eprintln!("connection error: {}", e);
+        }
+    });
+
+    // form replication connection
+    let (rclient, rconnection) = connect_replication(conninfo, NoTls, ReplicationMode::Physical)
+        .await
+        .unwrap();
+    tokio::spawn(async move {
+        if let Err(e) = rconnection.await {
+            eprintln!("connection error: {}", e);
+        }
+    });
+
+    let identify_system = rclient.identify_system().await.unwrap();
+    assert_eq!(identify_system.dbname(), None);
+    let show_port = rclient.show("port").await.unwrap();
+    assert_eq!(show_port, "5433");
+
+    let physical_stream = rclient
+        .start_physical_replication(None, identify_system.xlogpos(), None)
+        .await
+        .unwrap();
+
+    let _nrows = sclient
+        .execute("drop table test_physical_replication", &[])
+        .await
+        .unwrap();
+    let _nrows = sclient
+        .execute("create table test_physical_replication(i int)", &[])
+        .await
+        .unwrap();
+    let _nrows = sclient
+        .execute("insert into test_physical_replication values(1)", &[])
+        .await
+        .unwrap();
+
+    tokio::pin!(physical_stream);
+
+    let mut got_xlogdata = false;
+    while let Some(replication_message) = physical_stream.next().await {
+        if let ReplicationMessage::XLogData(_) = replication_message.unwrap() {
+            got_xlogdata = true;
+            break;
+        }
+    }
+
+    assert!(got_xlogdata);
+}