feat(postgres): add more backend protocol messages

sqlx-postgres/src/error/client.rs

@@ -12,6 +12,7 @@ pub enum PgClientError {
     NotUtf8(Utf8Error),
     UnknownAuthenticationMethod(u32),
     UnknownMessageType(u8),
+    UnknownTransactionStatus(u8),
     UnexpectedMessageType { ty: u8, context: &'static str },
 }
 
@@ -24,6 +25,10 @@ impl Display for PgClientError {
                 write!(f, "unknown authentication method: {}", method)
             }
 
+            Self::UnknownTransactionStatus(status) => {
+                write!(f, "in ReadyForQuery, unknown transaction status: {}", status)
+            }
+
             Self::UnknownMessageType(ty) => {
                 write!(f, "unknown protocol message type: '{}' ({})", *ty as char, *ty)
             }

sqlx-postgres/src/protocol/backend.rs

@@ -1,7 +1,19 @@
 mod auth;
+mod data_row;
+mod key_data;
 mod message;
+mod parameter_description;
+mod parameter_status;
+mod ready_for_query;
+mod row_description;
 mod sasl;
 
 pub(crate) use auth::{Authentication, AuthenticationMd5Password};
+pub(crate) use data_row::DataRow;
+pub(crate) use key_data::KeyData;
 pub(crate) use message::{BackendMessage, BackendMessageType};
+pub(crate) use parameter_description::ParameterDescription;
+pub(crate) use parameter_status::ParameterStatus;
+pub(crate) use ready_for_query::ReadyForQuery;
+pub(crate) use row_description::RowDescription;
 pub(crate) use sasl::{AuthenticationSasl, AuthenticationSaslContinue, AuthenticationSaslFinal};

sqlx-postgres/src/protocol/backend/data_row.rs

@@ -0,0 +1,31 @@
+use bytes::{Buf, Bytes};
+use sqlx_core::io::Deserialize;
+use sqlx_core::Result;
+
+#[derive(Debug)]
+pub(crate) struct DataRow {
+    pub(crate) values: Vec<Option<Bytes>>,
+}
+
+impl Deserialize<'_> for DataRow {
+    fn deserialize_with(mut buf: Bytes, _: ()) -> Result<Self> {
+        let cnt = buf.get_u16() as usize;
+
+        let mut values = Vec::with_capacity(cnt);
+
+        for _ in 0..cnt {
+            // length of the column value, in bytes (this count does not include itself)
+            // can be zero. as a special case, -1 indicates a NULL column value
+            // no value bytes follow in the NULL case
+            let length = buf.get_i32();
+
+            if length < 0 {
+                values.push(None);
+            } else {
+                values.push(Some(buf.split_to(length as usize)));
+            }
+        }
+
+        Ok(Self { values })
+    }
+}

sqlx-postgres/src/protocol/backend/key_data.rs

@@ -0,0 +1,36 @@
+use bytes::{Buf, Bytes};
+use sqlx_core::io::Deserialize;
+use sqlx_core::Result;
+
+#[derive(Debug)]
+pub(crate) struct KeyData {
+    /// The process ID of this database.
+    pub(crate) process_id: u32,
+
+    /// The secret key of this database.
+    pub(crate) secret_key: u32,
+}
+
+impl Deserialize<'_> for KeyData {
+    fn deserialize_with(mut buf: Bytes, _: ()) -> Result<Self> {
+        let process_id = buf.get_u32();
+        let secret_key = buf.get_u32();
+
+        Ok(Self { process_id, secret_key })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::KeyData;
+    use bytes::Bytes;
+    use sqlx_core::io::Deserialize;
+
+    #[test]
+    fn should_deserialize() {
+        let m = KeyData::deserialize(Bytes::from_static(b"\0\0'\xc6\x89R\xc5+")).unwrap();
+
+        assert_eq!(m.process_id, 10182);
+        assert_eq!(m.secret_key, 2303903019);
+    }
+}

sqlx-postgres/src/protocol/backend/parameter_description.rs

@@ -0,0 +1,21 @@
+use bytes::{Buf, Bytes};
+use sqlx_core::io::Deserialize;
+use sqlx_core::Result;
+
+#[derive(Debug)]
+pub(crate) struct ParameterDescription {
+    pub(crate) parameters: Vec<u32>,
+}
+
+impl Deserialize<'_> for ParameterDescription {
+    fn deserialize_with(mut buf: Bytes, _: ()) -> Result<Self> {
+        let cnt = buf.get_u16() as usize;
+        let mut parameters = Vec::with_capacity(cnt as usize);
+
+        for _ in 0..cnt {
+            parameters.push(buf.get_u32());
+        }
+
+        Ok(Self { parameters })
+    }
+}

sqlx-postgres/src/protocol/backend/parameter_status.rs

@@ -0,0 +1,19 @@
+use bytes::{Buf, Bytes};
+use bytestring::ByteString;
+use sqlx_core::io::{BufExt, Deserialize};
+use sqlx_core::Result;
+
+#[derive(Debug)]
+pub(crate) struct ParameterStatus {
+    pub(crate) name: ByteString,
+    pub(crate) value: ByteString,
+}
+
+impl Deserialize<'_> for ParameterStatus {
+    fn deserialize_with(mut buf: Bytes, _: ()) -> Result<Self> {
+        let name = buf.get_str_nul()?;
+        let value = buf.get_str_nul()?;
+
+        Ok(Self { name, value })
+    }
+}

sqlx-postgres/src/protocol/backend/ready_for_query.rs

@@ -0,0 +1,38 @@
+use crate::PgClientError;
+use bytes::{Buf, Bytes};
+use sqlx_core::io::Deserialize;
+use sqlx_core::{Error, Result};
+
+#[derive(Debug)]
+#[repr(u8)]
+pub(crate) enum TransactionStatus {
+    /// Not in a transaction block.
+    Idle = b'I',
+
+    /// In a transaction block.
+    Transaction = b'T',
+
+    /// In a _failed_ transaction block. Queries will be rejected until block is ended.
+    Error = b'E',
+}
+
+#[derive(Debug)]
+pub(crate) struct ReadyForQuery {
+    pub(crate) transaction_status: TransactionStatus,
+}
+
+impl Deserialize<'_> for ReadyForQuery {
+    fn deserialize_with(buf: Bytes, _: ()) -> Result<Self> {
+        let status = match buf[0] {
+            b'I' => TransactionStatus::Idle,
+            b'T' => TransactionStatus::Transaction,
+            b'E' => TransactionStatus::Error,
+
+            status => {
+                return Err(Error::client(PgClientError::UnknownTransactionStatus(status)));
+            }
+        };
+
+        Ok(Self { transaction_status: status })
+    }
+}

sqlx-postgres/src/protocol/backend/row_description.rs

@@ -0,0 +1,67 @@
+use bytes::{Buf, Bytes};
+use bytestring::ByteString;
+use sqlx_core::io::{BufExt, Deserialize};
+use sqlx_core::Result;
+use std::num::{NonZeroI16, NonZeroI32};
+
+#[derive(Debug)]
+pub(crate) struct RowDescription {
+    pub(crate) fields: Vec<Field>,
+}
+
+#[derive(Debug)]
+pub(crate) struct Field {
+    /// The name of the field.
+    pub(crate) name: ByteString,
+
+    /// If the field can be identified as a column of a specific table, the
+    /// object ID of the table; otherwise zero.
+    pub(crate) relation_id: Option<NonZeroI32>,
+
+    /// If the field can be identified as a column of a specific table, the attribute number of
+    /// the column; otherwise zero.
+    pub(crate) relation_attribute_no: Option<NonZeroI16>,
+
+    /// The object ID of the field's data type.
+    pub(crate) data_type_id: u32,
+
+    /// The data type size (see pg_type.typlen). Note that negative values denote
+    /// variable-width types.
+    pub(crate) data_type_size: i16,
+
+    /// The type modifier (see pg_attribute.atttypmod). The meaning of the
+    /// modifier is type-specific.
+    pub(crate) type_modifier: i32,
+
+    /// The format code being used for the field.
+    pub(crate) format: i16,
+}
+
+impl Deserialize<'_> for RowDescription {
+    fn deserialize_with(mut buf: Bytes, _: ()) -> Result<Self> {
+        let cnt = buf.get_u16() as usize;
+        let mut fields = Vec::with_capacity(cnt);
+
+        for _ in 0..cnt {
+            let name = buf.get_str_nul()?;
+            let relation_id = buf.get_i32();
+            let relation_attribute_no = buf.get_i16();
+            let data_type_id = buf.get_u32();
+            let data_type_size = buf.get_i16();
+            let type_modifier = buf.get_i32();
+            let format = buf.get_i16();
+
+            fields.push(Field {
+                name,
+                relation_id: NonZeroI32::new(relation_id),
+                relation_attribute_no: NonZeroI16::new(relation_attribute_no),
+                data_type_id,
+                data_type_size,
+                type_modifier,
+                format,
+            })
+        }
+
+        Ok(Self { fields })
+    }
+}