Explicitly terminate the connection in sync API

Closes #613

Author: sfackler

postgres/src/client.rs

@@ -3,6 +3,7 @@ use crate::{
     CancelToken, Config, CopyInWriter, CopyOutReader, Notifications, RowIter, Statement,
     ToStatement, Transaction, TransactionBuilder,
 };
+use std::task::Poll;
 use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
 use tokio_postgres::types::{ToSql, Type};
 use tokio_postgres::{Error, Row, SimpleQueryMessage, Socket};
@@ -13,6 +14,12 @@ pub struct Client {
     client: tokio_postgres::Client,
 }
 
+impl Drop for Client {
+    fn drop(&mut self) {
+        let _ = self.close_inner();
+    }
+}
+
 impl Client {
     pub(crate) fn new(connection: Connection, client: tokio_postgres::Client) -> Client {
         Client { connection, client }
@@ -524,4 +531,24 @@ impl Client {
     pub fn is_closed(&self) -> bool {
         self.client.is_closed()
     }
+
+    /// Closes the client's connection to the server.
+    ///
+    /// This is equivalent to `Client`'s `Drop` implementation, except that it returns any error encountered to the
+    /// caller.
+    pub fn close(mut self) -> Result<(), Error> {
+        self.close_inner()
+    }
+
+    fn close_inner(&mut self) -> Result<(), Error> {
+        self.client.__private_api_close();
+
+        self.connection.poll_block_on(|_, _, done| {
+            if done {
+                Poll::Ready(Ok(()))
+            } else {
+                Poll::Pending
+            }
+        })
+    }
 }

postgres/src/test.rs

@@ -475,3 +475,9 @@ fn notifications_timeout_iter() {
     assert_eq!(notifications[0].payload(), "hello");
     assert_eq!(notifications[1].payload(), "world");
 }
+
+#[test]
+fn explicit_close() {
+    let client = Client::connect("host=localhost port=5433 user=postgres", NoTls).unwrap();
+    client.close().unwrap();
+}

tokio-postgres/src/client.rs

@@ -529,6 +529,11 @@ impl Client {
     pub fn is_closed(&self) -> bool {
         self.inner.sender.is_closed()
     }
+
+    #[doc(hidden)]
+    pub fn __private_api_close(&mut self) {
+        self.inner.sender.close_channel()
+    }
 }
 
 impl fmt::Debug for Client {