implement SHOW command

Author: jeff-davis

tokio-postgres/src/replication_client.rs

@@ -1,14 +1,16 @@
 use crate::client::{InnerClient, Responses};
 use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
-use crate::types::Lsn;
+use crate::types::{Lsn, Type};
 use crate::{simple_query, Client, Error};
 use bytes::Bytes;
+use fallible_iterator::FallibleIterator;
 use futures::{ready, Stream};
 use pin_project_lite::pin_project;
-use postgres_protocol::message::backend::Message;
-use postgres_protocol::message::backend::ReplicationMessage;
+use postgres_protocol::message::backend::{Field, Message, ReplicationMessage};
+use std::io;
 use std::marker::PhantomPinned;
+use std::ops::Range;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -19,6 +21,58 @@ use std::task::{Context, Poll};
 pub struct ReplicationClient(pub(crate) Client);
 
 impl ReplicationClient {
+    /// show the value of the given setting
+    pub async fn show(&self, name: &str) -> Result<Vec<u8>, Error> {
+        let iclient = self.0.inner();
+        let command = format!("SHOW \"{}\"", name);
+        let buf = simple_query::encode(iclient, &command)?;
+        let mut responses = iclient.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+        match responses.next().await? {
+            Message::RowDescription(rowdesc) => {
+                let fields: Vec<Field<'_>> = rowdesc.fields().collect().map_err(Error::parse)?;
+                if fields.len() != 1 {
+                    return Err(Error::parse(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "expected single column in response",
+                    )))
+                }
+                if fields[0].type_oid() != Type::TEXT.oid() {
+                    return Err(Error::parse(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "expected single text column in response",
+                    )))
+                }
+            }
+            _ => return Err(Error::unexpected_message()),
+        };
+
+        match responses.next().await? {
+            Message::DataRow(d) => {
+                let ranges: Vec<Option<Range<usize>>> = d.ranges().collect().map_err(Error::parse)?;
+                if ranges.len() != 1 {
+                    return Err(Error::parse(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "expected single column in response",
+                    )))
+                }
+                // fetch only column
+                match &ranges[0] {
+                    Some(r) => Ok(Vec::from(&d.buffer()[r.to_owned()])),
+                    None => {
+                        Err(Error::parse(io::Error::new(
+                            io::ErrorKind::InvalidInput,
+                            "unexpected NULL setting in response",
+                        )))
+                    }
+                }
+            }
+            m => {
+                dbg!(m);
+                Err(Error::unexpected_message())
+            }
+        }
+    }
+
     /// Begin physical replication, consuming the replication client and producing a replication stream.
     ///
     /// Replication begins starting with the given Log Sequence Number
@@ -31,7 +85,7 @@ impl ReplicationClient {
     ) -> Result<ReplicationStream, Error> {
         let iclient = self.0.inner();
         let slot = match slot_name {
-            Some(name) => format!(" SLOT {}", name),
+            Some(name) => format!(" SLOT \"{}\"", name),
             None => String::from(""),
         };
         let timeline = match timeline_id {
@@ -70,7 +124,7 @@ impl ReplicationClient {
                 " ({})",
                 options
                     .iter()
-                    .map(|pair| format!("{} '{}'", pair.0, pair.1))
+                    .map(|pair| format!("\"{}\" '{}'", pair.0, pair.1))
                     .collect::<Vec<String>>()
                     .as_slice()
                     .join(", ")