Add new postgres manager builder and support a custom validation query

Issue djc#73 was looking to make a "DISCARD ALL" query possible when a
connection is returned to the pool. Because async drop isn't a thing,
this isn't really possible at the moment. However, we can use the
`is_valid` method on the `ManagedConnection` trait to discard all
session state before yielding the connection.

This PR makes it possible to change the query used within the `is_valid`
method call in `bb8_postgres`.

To make configuring the validation query a little easier, I added a new
postgres connection manager builder helper type. Let me know if you're
:+1: or :-1: on that.

This also includes an example called buidler.rs that shows how someone
could use `DISCARD ALL` as a validation query. It then prints a few
things to show that session state was indeed cleared between checkouts.

```
$ cargo run --example builder
...
The current connection PID: 86445
BB8 says, "beep boop"
The current connection PID: 86445
After DISCARD ALL on checkout, BB8 says, ""
```

Author: film42

postgres/examples/builder.rs

@@ -0,0 +1,78 @@
+use bb8::{Pool, PooledConnection, RunError};
+use bb8_postgres::{PostgresConnectionManager, PostgresConnectionManagerBuilder};
+
+use std::error::Error;
+use std::str::FromStr;
+
+// Demonstrate the postgres connection manager builder and its validation query.
+//
+// The simplest way to start the db is using Docker:
+// docker run --name gotham-middleware-postgres -e POSTGRES_PASSWORD=mysecretpassword -p 5432:5432 -d postgres
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let config =
+        tokio_postgres::config::Config::from_str("postgresql://postgres:docker@localhost:5432")
+            .unwrap();
+
+    let pg_mgr = PostgresConnectionManagerBuilder::new()
+        .tls(tokio_postgres::NoTls)
+        .validation_query("DISCARD ALL".into())
+        .config(config)
+        .build()?;
+
+    let pool = match Pool::builder().build(pg_mgr).await {
+        Ok(pool) => pool,
+        Err(e) => panic!("builder error: {:?}", e),
+    };
+
+    // The first checkout sets a session parameter.
+    {
+        let connection = pool.get().await?;
+        println!(
+            "The current connection PID: {}",
+            get_pid(&connection).await?
+        );
+        set_session_parameter(&connection, "bb8.greeting", "beep boop").await?;
+        let greeting = show_session_parameter(&connection, "bb8.greeting").await?;
+        println!("BB8 says, {:?}", greeting);
+    }
+
+    // The second checkout issues the DISCARD ALL validation query.
+    {
+        let connection = pool.get().await?;
+        println!(
+            "The current connection PID: {}",
+            get_pid(&connection).await?
+        );
+        let greeting = show_session_parameter(&connection, "bb8.greeting").await?;
+        println!("After DISCARD ALL on checkout, BB8 says, {:?}", greeting);
+    }
+
+    Ok(())
+}
+
+async fn get_pid(
+    connection: &PooledConnection<'_, PostgresConnectionManager<tokio_postgres::NoTls>>,
+) -> Result<i32, RunError<tokio_postgres::Error>> {
+    let row = connection.query_one("select pg_backend_pid()", &[]).await?;
+    Ok(row.get::<usize, i32>(0))
+}
+
+async fn set_session_parameter(
+    connection: &PooledConnection<'_, PostgresConnectionManager<tokio_postgres::NoTls>>,
+    key: &str,
+    value: &str,
+) -> Result<(), RunError<tokio_postgres::Error>> {
+    let query = format!("SET {} = '{}'", key, value);
+    connection.query(query.as_str(), &[]).await?;
+    Ok(())
+}
+
+async fn show_session_parameter(
+    connection: &PooledConnection<'_, PostgresConnectionManager<tokio_postgres::NoTls>>,
+    key: &str,
+) -> Result<String, RunError<tokio_postgres::Error>> {
+    let query = format!("SHOW {}", key);
+    let row = connection.query_one(query.as_str(), &[]).await?;
+    Ok(row.get::<usize, String>(0))
+}

postgres/src/lib.rs

@@ -9,8 +9,76 @@ use tokio_postgres::config::Config;
 use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
 use tokio_postgres::{Client, Error, Socket};
 
-use std::fmt;
 use std::str::FromStr;
+use std::{error, fmt};
+
+/// An error type for the postgres connection manager builder.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BuilderError(String);
+
+impl fmt::Display for BuilderError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "Builder error: {}", &self)
+    }
+}
+
+impl error::Error for BuilderError {}
+
+/// A builder for a postgres connection manager.
+#[derive(Debug)]
+pub struct PostgresConnectionManagerBuilder<Tls>
+where
+    Tls: MakeTlsConnect<Socket>,
+{
+    config: Option<Config>,
+    validation_query: String,
+    tls: Option<Tls>,
+}
+
+impl<Tls> PostgresConnectionManagerBuilder<Tls>
+where
+    Tls: MakeTlsConnect<Socket>,
+{
+    /// Create a new builder.
+    pub fn new() -> Self {
+        Self {
+            config: None,
+            validation_query: "".into(),
+            tls: None,
+        }
+    }
+
+    /// Set the postgres configuration.
+    pub fn config(mut self, config: Config) -> Self {
+        self.config = Some(config);
+        self
+    }
+
+    /// Set the validation query used by [`bb8::ManageConnection::is_valid`].
+    pub fn validation_query(mut self, query: String) -> Self {
+        self.validation_query = query;
+        self
+    }
+
+    /// Set the TLS configuraiton.
+    pub fn tls(mut self, tls: Tls) -> Self {
+        self.tls = Some(tls);
+        self
+    }
+
+    /// Build a postgres connection manager.
+    pub fn build(self) -> Result<PostgresConnectionManager<Tls>, Box<dyn std::error::Error>> {
+        Ok(PostgresConnectionManager {
+            tls: self.tls.ok_or_else(|| {
+                BuilderError("the tls config was set during the build process".into())
+            })?,
+            config: self.config.ok_or_else(|| {
+                BuilderError("the config was set during the build process".into())
+            })?,
+            validation_query: self.validation_query,
+        })
+    }
+}
 
 /// A `bb8::ManageConnection` for `tokio_postgres::Connection`s.
 #[derive(Clone)]
@@ -20,6 +88,7 @@ where
 {
     config: Config,
     tls: Tls,
+    validation_query: String,
 }
 
 impl<Tls> PostgresConnectionManager<Tls>
@@ -28,7 +97,11 @@ where
 {
     /// Create a new `PostgresConnectionManager` with the specified `config`.
     pub fn new(config: Config, tls: Tls) -> PostgresConnectionManager<Tls> {
-        PostgresConnectionManager { config, tls }
+        PostgresConnectionManager {
+            config,
+            tls,
+            validation_query: "".into(),
+        }
     }
 
     /// Create a new `PostgresConnectionManager`, parsing the config from `params`.
@@ -68,7 +141,8 @@ where
         &self,
         conn: &mut bb8::PooledConnection<'_, Self>,
     ) -> Result<(), Self::Error> {
-        conn.simple_query("").await.map(|_| ())
+        conn.simple_query(&self.validation_query).await?;
+        Ok(())
     }
 
     fn has_broken(&self, conn: &mut Self::Connection) -> bool {