standby status update

Author: jeff-davis

postgres-protocol/src/message/frontend.rs

@@ -134,19 +134,19 @@ pub fn close(variant: u8, name: &str, buf: &mut BytesMut) -> io::Result<()> {
 
 #[inline]
 pub fn standby_status_update(
-    write_lsn: i64,
-    flush_lsn: i64,
-    apply_lsn: i64,
+    write_lsn: u64,
+    flush_lsn: u64,
+    apply_lsn: u64,
     timestamp: i64,
     reply: u8,
     buf: &mut BytesMut,
 ) -> io::Result<()> {
     buf.put_u8(b'd');
     write_body(buf, |buf| {
         buf.put_u8(b'r');
-        buf.put_i64(write_lsn + 1);
-        buf.put_i64(flush_lsn + 1);
-        buf.put_i64(apply_lsn + 1);
+        buf.put_u64(write_lsn);
+        buf.put_u64(flush_lsn);
+        buf.put_u64(apply_lsn);
         buf.put_i64(timestamp);
         buf.put_u8(reply);
         Ok(())

tokio-postgres/src/replication_client.rs

@@ -3,12 +3,13 @@ use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
 use crate::types::{Lsn, Type};
 use crate::{simple_query, Client, Error};
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use fallible_iterator::FallibleIterator;
 use futures::{ready, Stream};
 use pin_project_lite::pin_project;
 use postgres_protocol::escape::{escape_identifier, escape_literal};
 use postgres_protocol::message::backend::{Message, ReplicationMessage};
+use postgres_protocol::message::frontend;
 use std::io;
 use std::marker::PhantomPinned;
 use std::pin::Pin;
@@ -213,6 +214,29 @@ impl ReplicationClient {
             _p: PhantomPinned,
         })
     }
+
+    /// Send update to server.
+    pub async fn standby_status_update(
+        &self,
+        write_lsn: Lsn,
+        flush_lsn: Lsn,
+        apply_lsn: Lsn,
+        ts: i64,
+        reply: u8,
+    ) -> Result<(), Error> {
+        let iclient = self.0.inner();
+        let mut buf = BytesMut::new();
+        let _ = frontend::standby_status_update(
+            write_lsn.into(),
+            flush_lsn.into(),
+            apply_lsn.into(),
+            ts as i64,
+            reply,
+            &mut buf,
+        );
+        let _ = iclient.send(RequestMessages::Single(FrontendMessage::Raw(buf.freeze())))?;
+        Ok(())
+    }
 }
 
 async fn start(client: &InnerClient, buf: Bytes) -> Result<Responses, Error> {