replication: fix integration tests

Signed-off-by: Petros Angelatos <[email protected]>

Author: petrosagg

tokio-postgres/tests/test/replication.rs

@@ -1,13 +1,11 @@
+use futures::StreamExt;
 use postgres_protocol::message::backend::ReplicationMessage;
-use tokio::stream::StreamExt;
+use postgres_protocol::replication::pgoutput::{LogicalReplicationMessage, PgOutput};
+use postgres_protocol::replication::DecodingPlugin;
 use tokio_postgres::replication_client::ReplicationClient;
 use tokio_postgres::Client;
 use tokio_postgres::{connect, connect_replication, NoTls, ReplicationMode};
 
-const LOGICAL_BEGIN_TAG: u8 = b'B';
-const LOGICAL_COMMIT_TAG: u8 = b'C';
-const LOGICAL_INSERT_TAG: u8 = b'I';
-
 // Tests missing for timeline_history(). For a timeline history to be
 // available, it requires a point-in-time-recovery or a standby
 // promotion; neither of which is done in the current test setup.
@@ -89,24 +87,20 @@ async fn logical_replication() {
     assert_eq!(identify_system.dbname().unwrap(), "postgres");
 
     let slot = "test_logical_slot";
-    let plugin = "pgoutput";
+    let plugin = PgOutput::new(vec!["test_logical_pub".to_string()]);
     let _ = rclient.drop_replication_slot(slot, false).await.unwrap();
     let slotdesc = rclient
-        .create_logical_replication_slot(slot, false, plugin, None)
+        .create_logical_replication_slot(slot, false, &plugin, None)
         .await
         .unwrap();
     assert_eq!(slotdesc.slot_name(), slot);
     assert!(slotdesc.snapshot_name().is_some());
-    assert_eq!(slotdesc.output_plugin(), Some(plugin));
+    assert_eq!(slotdesc.output_plugin(), Some(plugin.name()));
 
     let xlog_start = identify_system.xlogpos();
-    let options = &vec![
-        ("proto_version", "1"),
-        ("publication_names", "test_logical_pub"),
-    ];
 
     let mut logical_stream = rclient
-        .start_logical_replication(slot, xlog_start, options)
+        .start_logical_replication(slot, xlog_start, &plugin)
         .await
         .unwrap();
 
@@ -120,20 +114,21 @@ async fn logical_replication() {
     let mut got_commit = false;
     while let Some(replication_message) = logical_stream.next().await {
         if let ReplicationMessage::XLogData(msg) = replication_message.unwrap() {
-            match msg.data()[0] {
-                LOGICAL_BEGIN_TAG => {
+            use LogicalReplicationMessage::*;
+            match msg.data() {
+                Begin(_) => {
                     assert!(!got_begin);
                     assert!(!got_insert);
                     assert!(!got_commit);
                     got_begin = true;
                 }
-                LOGICAL_INSERT_TAG => {
+                Insert(_) => {
                     assert!(got_begin);
                     assert!(!got_insert);
                     assert!(!got_commit);
                     got_insert = true;
                 }
-                LOGICAL_COMMIT_TAG => {
+                Commit(_) => {
                     assert!(got_begin);
                     assert!(got_insert);
                     assert!(!got_commit);
@@ -153,10 +148,6 @@ async fn logical_replication() {
     simple_exec(&sclient, "drop publication if exists test_logical_pub").await;
 }
 
-// test for base backup
-#[tokio::test]
-async fn base_backup() {}
-
 // Test that a dropped replication stream properly returns to normal
 // command processing in the ReplicationClient.
 //