Add a new exec_simple_query method

This adds a new `exec_simple_query` method so we can make 'out of band'
queries to servers that don't interfere with pools at all.
In order to reuse startup code for making these simple queries,
we need to set the stats (`Reporter`) optional, so using these
simple queries wont interfere with stats.

Author: magec

src/server.rs

@@ -62,7 +62,7 @@ pub struct Server {
     connected_at: chrono::naive::NaiveDateTime,
 
     /// Reports various metrics, e.g. data sent & received.
-    stats: Reporter,
+    stats: Option<Reporter>,
 
     /// Application name using the server at the moment.
     application_name: String,
@@ -82,7 +82,7 @@ impl Server {
         user: &User,
         database: &str,
         client_server_map: ClientServerMap,
-        stats: Reporter,
+        stats: Option<Reporter>,
     ) -> Result<Server, Error> {
         let mut stream =
             match TcpStream::connect(&format!("{}:{}", &address.host, address.port)).await {
@@ -396,7 +396,10 @@ impl Server {
     /// Send messages to the server from the client.
     pub async fn send(&mut self, messages: &BytesMut) -> Result<(), Error> {
         self.mirror_send(messages);
-        self.stats.data_sent(messages.len(), self.server_id);
+
+        if let Some(stats) = &self.stats {
+	    stats.data_sent(messages.len(), self.server_id);
+        }
 
         match write_all_half(&mut self.write, messages).await {
             Ok(_) => {
@@ -545,7 +548,9 @@ impl Server {
         let bytes = self.buffer.clone();
 
         // Keep track of how much data we got from the server for stats.
-        self.stats.data_received(bytes.len(), self.server_id);
+        if let Some(stats) = &self.stats {
+            stats.data_received(bytes.len(), self.server_id);
+        }
 
         // Clear the buffer for next query.
         self.buffer.clear();
@@ -700,6 +705,106 @@ impl Server {
             None => (),
         }
     }
+
+    // This is so we can execute out of band queries to the server.
+    // The connection will be opened, the query executed and closed.
+    pub async fn exec_simple_query(
+        address: &Address,
+        user: &User,
+        query: &str,
+    ) -> Result<Vec<String>, Error> {
+        let client_server_map: ClientServerMap = Arc::new(Mutex::new(HashMap::new()));
+
+        debug!("Connecting to server to obtain auth hashes.");
+        let mut server = Server::startup(
+            0,
+            address,
+            user,
+            &address.database,
+            client_server_map,
+            None,
+            Arc::new(RwLock::new(None)),
+        )
+        .await?;
+        debug!("Connected!, sending query.");
+        server.send(&simple_query(query)).await?;
+        let mut message = server.recv().await?;
+
+        Ok(parse_query_message(&mut message).await?)
+    }
+}
+
+async fn parse_query_message(message: &mut BytesMut) -> Result<Vec<String>, Error> {
+    let mut pair = Vec::<String>::new();
+    match message::backend::Message::parse(message) {
+        Ok(Some(message::backend::Message::RowDescription(_description))) => {}
+        Ok(Some(message::backend::Message::ErrorResponse(err))) => {
+            return Err(Error::ProtocolSyncError(format!(
+                "Protocol error parsing response. Err: {:?}",
+                err.fields()
+                    .iterator()
+                    .fold(String::default(), |acc, element| acc
+                        + element.unwrap().value())
+            )))
+        }
+        Ok(_) => {
+            return Err(Error::ProtocolSyncError(
+                "Protocol error, expected Row Description.".to_string(),
+            ))
+        }
+        Err(err) => {
+            return Err(Error::ProtocolSyncError(format!(
+                "Protocol error parsing response. Err: {:?}",
+                err
+            )))
+        }
+    }
+
+    while !message.is_empty() {
+        match message::backend::Message::parse(message) {
+            Ok(postgres_message) => {
+                match postgres_message {
+                    Some(message::backend::Message::DataRow(data)) => {
+                        let buf = data.buffer();
+                        trace!("Data: {:?}", buf);
+
+                        for item in data.ranges().iterator() {
+                            match item.as_ref() {
+                            Ok(range) => match range {
+                                Some(range) => {
+                                    pair.push(String::from_utf8_lossy(&buf[range.clone()]).to_string());
+                                }
+                                None => return Err(Error::ProtocolSyncError(String::from(
+                                    "Data expected while receiving query auth data, found nothing.",
+                                ))),
+                            },
+                            Err(err) => {
+                                return Err(Error::ProtocolSyncError(format!(
+                                    "Data error, err: {:?}",
+                                    err
+                                )))
+                            }
+                            }
+                        }
+                    }
+                    Some(message::backend::Message::CommandComplete(_)) => {}
+                    Some(message::backend::Message::ReadyForQuery(_)) => {}
+                    _ => {
+                        return Err(Error::ProtocolSyncError(
+                            "Unexpected message while receiving auth query data.".to_string(),
+                        ))
+                    }
+                }
+            }
+            Err(err) => {
+                return Err(Error::ProtocolSyncError(format!(
+                    "Parse error, err: {:?}",
+                    err
+                )))
+            }
+        };
+    }
+    Ok(pair)
 }
 
 impl Drop for Server {
@@ -708,7 +813,10 @@ impl Drop for Server {
     /// for a write.
     fn drop(&mut self) {
         self.mirror_disconnect();
-        self.stats.server_disconnecting(self.server_id);
+
+        if let Some(stats) = &self.stats {
+            stats.server_disconnecting(self.server_id);
+        }
 
         let mut bytes = BytesMut::with_capacity(4);
         bytes.put_u8(b'X');