Add support for CopyBoth mode

sqlx-postgres/Cargo.toml

@@ -50,6 +50,7 @@ base64 = { version = "0.21.0", default-features = false, features = ["std"] }
 bitflags = { version = "2", default-features = false }
 byteorder = { version = "1.4.3", default-features = false, features = ["std"] }
 dotenvy = { workspace = true }
+flume = { version = "0.11.0", default-features = false, features = ["async"] }
 hex = "0.4.3"
 home = "0.5.5"
 itoa = "1.0.1"
@@ -71,5 +72,9 @@ workspace = true
 # We use JSON in the driver implementation itself so there's no reason not to enable it here.
 features = ["json"]
 
+
+[dev-dependencies]
+sqlx = { workspace = true, features = ["postgres"] }
+
 [target.'cfg(target_os = "windows")'.dependencies]
 etcetera = "0.8.0"

sqlx-postgres/src/advisory_lock.rs

@@ -98,7 +98,6 @@ impl PgAdvisoryLock {
     /// [hkdf]: https://datatracker.ietf.org/doc/html/rfc5869
     /// ### Example
     /// ```rust
-    /// # extern crate sqlx_core as sqlx;
     /// use sqlx::postgres::{PgAdvisoryLock, PgAdvisoryLockKey};
     ///
     /// let lock = PgAdvisoryLock::new("my first Postgres advisory lock!");

sqlx-postgres/src/connection/establish.rs

@@ -8,7 +8,7 @@ use crate::message::{
     Authentication, BackendKeyData, MessageFormat, Password, ReadyForQuery, Startup,
 };
 use crate::types::Oid;
-use crate::{PgConnectOptions, PgConnection};
+use crate::{PgConnectOptions, PgConnection, PgReplicationMode};
 
 // https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.3
 // https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.11
@@ -44,6 +44,13 @@ impl PgConnection {
             params.push(("options", options));
         }
 
+        if let Some(replication_mode) = options.replication_mode {
+            match replication_mode {
+                PgReplicationMode::Physical => params.push(("replication", "true")),
+                PgReplicationMode::Logical => params.push(("replication", "database")),
+            }
+        }
+
         stream
             .send(Startup {
                 username: Some(&options.username),

sqlx-postgres/src/lib.rs

@@ -17,6 +17,7 @@ mod listener;
 mod message;
 mod options;
 mod query_result;
+mod replication;
 mod row;
 mod statement;
 mod transaction;
@@ -44,8 +45,9 @@ pub use database::Postgres;
 pub use error::{PgDatabaseError, PgErrorPosition};
 pub use listener::{PgListener, PgNotification};
 pub use message::PgSeverity;
-pub use options::{PgConnectOptions, PgSslMode};
+pub use options::{PgConnectOptions, PgReplicationMode, PgSslMode};
 pub use query_result::PgQueryResult;
+pub use replication::{PgCopyBothReceiver, PgCopyBothSender, PgReplication, PgReplicationPool};
 pub use row::PgRow;
 pub use statement::PgStatement;
 pub use transaction::PgTransactionManager;

sqlx-postgres/src/listener.rs

@@ -188,8 +188,8 @@ impl PgListener {
     /// # Example
     ///
     /// ```rust,no_run
-    /// # use sqlx_core::postgres::PgListener;
-    /// # use sqlx_core::error::Error;
+    /// # use sqlx::postgres::PgListener;
+    /// # use sqlx::error::Error;
     /// #
     /// # #[cfg(feature = "_rt")]
     /// # sqlx::__rt::test_block_on(async move {
@@ -219,8 +219,8 @@ impl PgListener {
     /// # Example
     ///
     /// ```rust,no_run
-    /// # use sqlx_core::postgres::PgListener;
-    /// # use sqlx_core::error::Error;
+    /// # use sqlx::postgres::PgListener;
+    /// # use sqlx::error::Error;
     /// #
     /// # #[cfg(feature = "_rt")]
     /// # sqlx::__rt::test_block_on(async move {

sqlx-postgres/src/message/copy.rs

@@ -3,7 +3,7 @@ use crate::io::{BufExt, BufMutExt, Decode, Encode};
 use sqlx_core::bytes::{Buf, BufMut, Bytes};
 use std::ops::Deref;
 
-/// The same structure is sent for both `CopyInResponse` and `CopyOutResponse`
+/// The same structure is sent for `CopyInResponse`, `CopyOutResponse` and `CopyBothResponse`.
 pub struct CopyResponse {
     pub format: i8,
     pub num_columns: i16,

sqlx-postgres/src/message/mod.rs

@@ -64,6 +64,7 @@ pub enum MessageFormat {
     CommandComplete,
     CopyData,
     CopyDone,
+    CopyBothResponse,
     CopyInResponse,
     CopyOutResponse,
     DataRow,
@@ -118,6 +119,7 @@ impl MessageFormat {
             b'R' => MessageFormat::Authentication,
             b'S' => MessageFormat::ParameterStatus,
             b'T' => MessageFormat::RowDescription,
+            b'W' => MessageFormat::CopyBothResponse,
             b'Z' => MessageFormat::ReadyForQuery,
             b'n' => MessageFormat::NoData,
             b's' => MessageFormat::PortalSuspended,