support timeline_history

Author: jeff-davis

tokio-postgres/src/replication_client.rs

@@ -41,6 +41,22 @@ impl IdentifySystem {
     }
 }
 
+#[derive(Debug)]
+pub struct TimelineHistory {
+    filename: Vec<u8>,
+    content: Vec<u8>,
+}
+
+impl TimelineHistory {
+    pub fn filename(&self) -> &[u8] {
+        &self.content
+    }
+
+    pub fn content(&self) -> &[u8] {
+        &self.content
+    }
+}
+
 /// Replication client connection.
 ///
 /// A replication client is used to issue replication commands, begin
@@ -77,9 +93,13 @@ impl ReplicationClient {
 
         assert_eq!(fields.len(), 4);
         assert_eq!(fields[0].type_oid(), Type::TEXT.oid());
+        assert_eq!(fields[0].format(), 0);
         assert_eq!(fields[1].type_oid(), Type::INT4.oid());
+        assert_eq!(fields[1].format(), 0);
         assert_eq!(fields[2].type_oid(), Type::TEXT.oid());
+        assert_eq!(fields[2].format(), 0);
         assert_eq!(fields[3].type_oid(), Type::TEXT.oid());
+        assert_eq!(fields[3].format(), 0);
         assert_eq!(ranges.len(), 4);
 
         let values: Vec<Option<&str>> = ranges
@@ -92,7 +112,7 @@ impl ReplicationClient {
             .collect::<Vec<_>>();
 
         Ok(IdentifySystem {
-            systemid: String::from(values[0].unwrap()),
+            systemid: values[0].unwrap().to_string(),
             timeline: values[1].unwrap().parse::<u32>().unwrap(),
             xlogpos: Lsn::from(values[2].unwrap()),
             dbname: values[3].map(String::from),
@@ -135,6 +155,55 @@ impl ReplicationClient {
         Ok(String::from(val))
     }
 
+    /// show the value of the given setting
+    pub async fn timeline_history(&self, timeline_id: u32) -> Result<TimelineHistory, Error> {
+        let iclient = self.0.inner();
+        let command = format!("TIMELINE_HISTORY {}", timeline_id);
+        let buf = simple_query::encode(iclient, &command)?;
+        let mut responses = iclient.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+
+        let rowdesc = match responses.next().await? {
+            Message::RowDescription(m) => m,
+            _ => return Err(Error::unexpected_message()),
+        };
+        let datarow = match responses.next().await? {
+            Message::DataRow(m) => m,
+            _ => return Err(Error::unexpected_message()),
+        };
+        match responses.next().await? {
+            Message::CommandComplete(_) => (),
+            _ => return Err(Error::unexpected_message()),
+        };
+        match responses.next().await? {
+            Message::ReadyForQuery(_) => (),
+            _ => return Err(Error::unexpected_message()),
+        };
+
+        let fields = rowdesc.fields().collect::<Vec<_>>().map_err(Error::parse)?;
+        let ranges = datarow.ranges().collect::<Vec<_>>().map_err(Error::parse)?;
+
+        assert_eq!(fields.len(), 2);
+
+        // Both fields should actually be labeled BYTEA with format
+        // code 1 (for binary), because they return raw
+        // bytes. Unfortunately, postgres returns a misleading
+        // descriptor.
+        //assert_eq!(fields[0].type_oid(), Type::BYTEA.oid());
+        //assert_eq!(fields[0].format(), 1);
+        //assert_eq!(fields[1].type_oid(), Type::BYTEA.oid());
+        //assert_eq!(fields[1].format(), 1);
+
+        assert_eq!(ranges.len(), 2);
+
+        let filename = &datarow.buffer()[ranges[0].to_owned().unwrap()];
+        let content = &datarow.buffer()[ranges[1].to_owned().unwrap()];
+
+        Ok(TimelineHistory {
+            filename: filename.to_owned(),
+            content: content.to_owned(),
+        })
+    }
+
     /// Begin physical replication, consuming the replication client and producing a replication stream.
     ///
     /// Replication begins starting with the given Log Sequence Number

tokio-postgres/tests/test/replication.rs

@@ -8,10 +8,13 @@ const LOGICAL_BEGIN_TAG: u8 = b'B';
 const LOGICAL_COMMIT_TAG: u8 = b'C';
 const LOGICAL_INSERT_TAG: u8 = b'I';
 
+// Tests missing for timeline_history(). For a timeline history to be
+// available, it requires a point-in-time-recovery or a standby
+// promotion; neither of which is done in the current test setup.
+
 // test for:
 //   - identify_system
 //   - show
-//   X timeline history
 //   - physical replication
 #[tokio::test]
 async fn physical_replication() {