Add InnerClient::unpipelined_send() method.

Needed for CopyBoth (streaming replication) mode, where the client may
send new messages that are part of the existing request.

Author: jeff-davis

tokio-postgres/src/client.rs

@@ -70,7 +70,7 @@ pub struct InnerClient {
 impl InnerClient {
     pub fn send(&self, messages: RequestMessages) -> Result<Responses, Error> {
         let (sender, receiver) = mpsc::channel(1);
-        let request = Request { messages, sender };
+        let request = Request { messages: messages, sender: Some(sender) };
         self.sender
             .unbounded_send(request)
             .map_err(|_| Error::closed())?;
@@ -81,6 +81,18 @@ impl InnerClient {
         })
     }
 
+    // Send a message for the existing entry in the pipeline; don't
+    // create a new entry in the pipeline. This is needed for CopyBoth
+    // mode (i.e. streaming replication), where the client may send a
+    // new message that is part of the existing request.
+    pub fn unpipelined_send(&self, messages: RequestMessages) -> Result<(), Error> {
+        let request = Request { messages: messages, sender: None };
+        self.sender
+            .unbounded_send(request)
+            .map_err(|_| Error::closed())?;
+        Ok(())
+    }
+
     pub fn typeinfo(&self) -> Option<Statement> {
         self.state.lock().typeinfo.clone()
     }

tokio-postgres/src/connection.rs

@@ -25,7 +25,7 @@ pub enum RequestMessages {
 
 pub struct Request {
     pub messages: RequestMessages,
-    pub sender: mpsc::Sender<BackendMessages>,
+    pub sender: Option<mpsc::Sender<BackendMessages>>,
 }
 
 pub struct Response {
@@ -183,9 +183,11 @@ where
         match self.receiver.poll_next_unpin(cx) {
             Poll::Ready(Some(request)) => {
                 trace!("polled new request");
-                self.responses.push_back(Response {
-                    sender: request.sender,
-                });
+                if let Some(sender) = request.sender {
+                    self.responses.push_back(Response {
+                        sender: sender,
+                    });
+                }
                 Poll::Ready(Some(request.messages))
             }
             Poll::Ready(None) => Poll::Ready(None),