replication: fix documentation tests

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

Author: petrosagg

tokio-postgres/src/replication_client.rs

@@ -14,10 +14,8 @@
 //!
 //! # Physical Replication Client Example
 //! ```no_run
-//! extern crate tokio;
-//!
 //! use postgres_protocol::message::backend::ReplicationMessage;
-//! use tokio::stream::StreamExt;
+//! use futures::StreamExt;
 //! use tokio_postgres::{connect_replication, Error, NoTls, ReplicationMode};
 //!
 //! #[tokio::main]
@@ -61,11 +59,10 @@
 //! extension](https://github.com/eulerto/wal2json).
 //!
 //! ```no_run
-//! extern crate tokio;
-//!
 //! use postgres_protocol::message::backend::ReplicationMessage;
-//! use tokio::stream::StreamExt;
+//! use futures::StreamExt;
 //! use tokio_postgres::{connect_replication, Error, NoTls, ReplicationMode};
+//! use postgres_protocol::replication::pgoutput::{PgOutput, LogicalReplicationMessage};
 //!
 //! #[tokio::main]
 //! async fn main() -> Result<(), Error> {
@@ -85,23 +82,27 @@
 //!     let identify_system = rclient.identify_system().await?;
 //!
 //!     let slot = "my_slot";
-//!     let plugin = "wal2json";
-//!     let options = &vec![("pretty-print", "1")];
+//!     let plugin = PgOutput::new(vec!["publication_name".to_string()]);
 //!
 //!     let _slotdesc = rclient
-//!         .create_logical_replication_slot(slot, false, plugin, None)
+//!         .create_logical_replication_slot(slot, false, &plugin, None)
 //!         .await?;
 //!
-//!     let mut physical_stream = rclient
-//!         .start_logical_replication(slot, identify_system.xlogpos(), options)
+//!     let mut logical_stream = rclient
+//!         .start_logical_replication(slot, identify_system.xlogpos(), &plugin)
 //!         .await?;
 //!
-//!     while let Some(replication_message) = physical_stream.next().await {
+//!     while let Some(replication_message) = logical_stream.next().await {
 //!         match replication_message? {
 //!             ReplicationMessage::XLogData(xlog_data) => {
+//!                 use LogicalReplicationMessage::*;
 //!                 eprintln!("received XLogData: {:#?}", xlog_data);
-//!                 let json = std::str::from_utf8(xlog_data.data()).unwrap();
-//!                 eprintln!("JSON text: {}", json);
+//!                 match xlog_data.data() {
+//!                     Insert(insert) => eprintln!("insert event: {:?}", insert),
+//!                     Update(update) => eprintln!("update event: {:?}", update),
+//!                     Delete(delete) => eprintln!("delete event: {:?}", delete),
+//!                     event => eprintln!("other {:?}", event),
+//!                 }
 //!             }
 //!             ReplicationMessage::PrimaryKeepAlive(keepalive) => {
 //!                 eprintln!("received PrimaryKeepAlive: {:#?}", keepalive);
@@ -739,7 +740,7 @@ impl ReplicationClient {
 /// and data messages will be in
 /// [CopyData](postgres_protocol::message::backend::Message::CopyData).
 ///
-/// Intended to be used with the [next()](tokio::stream::StreamExt::next) method.
+/// Intended to be used with the [next()](futures::StreamExt::next) method.
 ///
 /// If the timeline specified with
 /// [start_physical_replication()](ReplicationClient::start_physical_replication)