support unnamed statements

postgres-protocol/src/message/backend.rs

@@ -72,6 +72,7 @@ impl Header {
 }
 
 /// An enum representing Postgres backend messages.
+#[derive(Debug, PartialEq)]
 #[non_exhaustive]
 pub enum Message {
     AuthenticationCleartextPassword,
@@ -333,6 +334,7 @@ impl Read for Buffer {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct AuthenticationMd5PasswordBody {
     salt: [u8; 4],
 }
@@ -344,6 +346,7 @@ impl AuthenticationMd5PasswordBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct AuthenticationGssContinueBody(Bytes);
 
 impl AuthenticationGssContinueBody {
@@ -353,6 +356,7 @@ impl AuthenticationGssContinueBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct AuthenticationSaslBody(Bytes);
 
 impl AuthenticationSaslBody {
@@ -362,6 +366,7 @@ impl AuthenticationSaslBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct SaslMechanisms<'a>(&'a [u8]);
 
 impl<'a> FallibleIterator for SaslMechanisms<'a> {
@@ -387,6 +392,7 @@ impl<'a> FallibleIterator for SaslMechanisms<'a> {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct AuthenticationSaslContinueBody(Bytes);
 
 impl AuthenticationSaslContinueBody {
@@ -396,6 +402,7 @@ impl AuthenticationSaslContinueBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct AuthenticationSaslFinalBody(Bytes);
 
 impl AuthenticationSaslFinalBody {
@@ -405,6 +412,7 @@ impl AuthenticationSaslFinalBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct BackendKeyDataBody {
     process_id: i32,
     secret_key: i32,
@@ -422,6 +430,7 @@ impl BackendKeyDataBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct CommandCompleteBody {
     tag: Bytes,
 }
@@ -433,6 +442,7 @@ impl CommandCompleteBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct CopyDataBody {
     storage: Bytes,
 }
@@ -449,6 +459,7 @@ impl CopyDataBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct CopyInResponseBody {
     format: u8,
     len: u16,
@@ -470,6 +481,7 @@ impl CopyInResponseBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct ColumnFormats<'a> {
     buf: &'a [u8],
     remaining: u16,
@@ -503,6 +515,7 @@ impl<'a> FallibleIterator for ColumnFormats<'a> {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct CopyOutResponseBody {
     format: u8,
     len: u16,
@@ -524,7 +537,7 @@ impl CopyOutResponseBody {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq)]
 pub struct DataRowBody {
     storage: Bytes,
     len: u16,
@@ -599,6 +612,7 @@ impl<'a> FallibleIterator for DataRowRanges<'a> {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct ErrorResponseBody {
     storage: Bytes,
 }
@@ -657,6 +671,7 @@ impl<'a> ErrorField<'a> {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct NoticeResponseBody {
     storage: Bytes,
 }
@@ -668,6 +683,7 @@ impl NoticeResponseBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct NotificationResponseBody {
     process_id: i32,
     channel: Bytes,
@@ -691,6 +707,7 @@ impl NotificationResponseBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct ParameterDescriptionBody {
     storage: Bytes,
     len: u16,
@@ -706,6 +723,7 @@ impl ParameterDescriptionBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct Parameters<'a> {
     buf: &'a [u8],
     remaining: u16,
@@ -739,6 +757,7 @@ impl<'a> FallibleIterator for Parameters<'a> {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct ParameterStatusBody {
     name: Bytes,
     value: Bytes,
@@ -756,6 +775,7 @@ impl ParameterStatusBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct ReadyForQueryBody {
     status: u8,
 }
@@ -767,6 +787,7 @@ impl ReadyForQueryBody {
     }
 }
 
+#[derive(Debug, PartialEq)]
 pub struct RowDescriptionBody {
     storage: Bytes,
     len: u16,

tokio-postgres/Cargo.toml

@@ -59,7 +59,7 @@ postgres-types = { version = "0.2.5", path = "../postgres-types" }
 tokio = { version = "1.27", features = ["io-util"] }
 tokio-util = { version = "0.7", features = ["codec"] }
 rand = "0.8.5"
-whoami = "1.4.1"
+whoami = "1.4"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 socket2 = { version = "0.5", features = ["all"] }

tokio-postgres/src/bind.rs

@@ -31,7 +31,7 @@ where
 
     match responses.next().await? {
         Message::BindComplete => {}
-        _ => return Err(Error::unexpected_message()),
+        m => return Err(Error::unexpected_message(m)),
     }
 
     Ok(Portal::new(client, name, statement))

tokio-postgres/src/client.rs

@@ -231,7 +231,11 @@ impl Client {
         query: &str,
         parameter_types: &[Type],
     ) -> Result<Statement, Error> {
-        prepare::prepare(&self.inner, query, parameter_types).await
+        prepare::prepare(&self.inner, query, parameter_types, false).await
+    }
+
+    pub(crate) async fn prepare_unnamed(&self, query: &str) -> Result<Statement, Error> {
+        prepare::prepare(&self.inner, query, &[], true).await
     }
 
     /// Executes a statement, returning a vector of the resulting rows.

tokio-postgres/src/connect.rs

@@ -195,7 +195,7 @@ where
                     }
                 }
                 Some(_) => {}
-                None => return Err(Error::unexpected_message()),
+                None => return Err(Error::closed()),
             }
         }
     }

tokio-postgres/src/connect_raw.rs

@@ -190,14 +190,14 @@ where
             ))
         }
         Some(Message::ErrorResponse(body)) => return Err(Error::db(body)),
-        Some(_) => return Err(Error::unexpected_message()),
+        Some(m) => return Err(Error::unexpected_message(m)),
         None => return Err(Error::closed()),
     }
 
     match stream.try_next().await.map_err(Error::io)? {
         Some(Message::AuthenticationOk) => Ok(()),
         Some(Message::ErrorResponse(body)) => Err(Error::db(body)),
-        Some(_) => Err(Error::unexpected_message()),
+        Some(m) => Err(Error::unexpected_message(m)),
         None => Err(Error::closed()),
     }
 }
@@ -291,7 +291,7 @@ where
     let body = match stream.try_next().await.map_err(Error::io)? {
         Some(Message::AuthenticationSaslContinue(body)) => body,
         Some(Message::ErrorResponse(body)) => return Err(Error::db(body)),
-        Some(_) => return Err(Error::unexpected_message()),
+        Some(m) => return Err(Error::unexpected_message(m)),
         None => return Err(Error::closed()),
     };
 
@@ -309,7 +309,7 @@ where
     let body = match stream.try_next().await.map_err(Error::io)? {
         Some(Message::AuthenticationSaslFinal(body)) => body,
         Some(Message::ErrorResponse(body)) => return Err(Error::db(body)),
-        Some(_) => return Err(Error::unexpected_message()),
+        Some(m) => return Err(Error::unexpected_message(m)),
         None => return Err(Error::closed()),
     };
 
@@ -348,7 +348,7 @@ where
             }
             Some(Message::ReadyForQuery(_)) => return Ok((process_id, secret_key, parameters)),
             Some(Message::ErrorResponse(body)) => return Err(Error::db(body)),
-            Some(_) => return Err(Error::unexpected_message()),
+            Some(m) => return Err(Error::unexpected_message(m)),
             None => return Err(Error::closed()),
         }
     }

tokio-postgres/src/connection.rs

@@ -139,7 +139,8 @@ where
                 Some(response) => response,
                 None => match messages.next().map_err(Error::parse)? {
                     Some(Message::ErrorResponse(error)) => return Err(Error::db(error)),
-                    _ => return Err(Error::unexpected_message()),
+                    Some(m) => return Err(Error::unexpected_message(m)),
+                    None => return Err(Error::closed()),
                 },
             };
 

tokio-postgres/src/copy_in.rs

@@ -114,7 +114,7 @@ where
                             let rows = extract_row_affected(&body)?;
                             return Poll::Ready(Ok(rows));
                         }
-                        _ => return Poll::Ready(Err(Error::unexpected_message())),
+                        m => return Poll::Ready(Err(Error::unexpected_message(m))),
                     }
                 }
             }
@@ -206,13 +206,19 @@ where
         .map_err(|_| Error::closed())?;
 
     match responses.next().await? {
+        Message::ParseComplete => {
+            match responses.next().await? {
+                Message::BindComplete => {}
+                m => return Err(Error::unexpected_message(m)),
+            }
+        }
         Message::BindComplete => {}
-        _ => return Err(Error::unexpected_message()),
+        m => return Err(Error::unexpected_message(m)),
     }
 
     match responses.next().await? {
         Message::CopyInResponse(_) => {}
-        _ => return Err(Error::unexpected_message()),
+        m => return Err(Error::unexpected_message(m)),
     }
 
     Ok(CopyInSink {

tokio-postgres/src/copy_out.rs

@@ -26,13 +26,17 @@ async fn start(client: &InnerClient, buf: Bytes) -> Result<Responses, Error> {
     let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 
     match responses.next().await? {
+        Message::ParseComplete => match responses.next().await? {
+            Message::BindComplete => {}
+            m => return Err(Error::unexpected_message(m)),
+        },
         Message::BindComplete => {}
-        _ => return Err(Error::unexpected_message()),
+        m => return Err(Error::unexpected_message(m)),
     }
 
     match responses.next().await? {
         Message::CopyOutResponse(_) => {}
-        _ => return Err(Error::unexpected_message()),
+        m => return Err(Error::unexpected_message(m)),
     }
 
     Ok(responses)
@@ -56,7 +60,7 @@ impl Stream for CopyOutStream {
         match ready!(this.responses.poll_next(cx)?) {
             Message::CopyData(body) => Poll::Ready(Some(Ok(body.into_bytes()))),
             Message::CopyDone => Poll::Ready(None),
-            _ => Poll::Ready(Some(Err(Error::unexpected_message()))),
+            m => Poll::Ready(Some(Err(Error::unexpected_message(m)))),
         }
     }
 }

tokio-postgres/src/error/mod.rs

@@ -1,7 +1,7 @@
 //! Errors.
 
 use fallible_iterator::FallibleIterator;
-use postgres_protocol::message::backend::{ErrorFields, ErrorResponseBody};
+use postgres_protocol::message::backend::{ErrorFields, ErrorResponseBody, Message};
 use std::error::{self, Error as _Error};
 use std::fmt;
 use std::io;
@@ -339,7 +339,7 @@ pub enum ErrorPosition {
 #[derive(Debug, PartialEq)]
 enum Kind {
     Io,
-    UnexpectedMessage,
+    UnexpectedMessage(Message),
     Tls,
     ToSql(usize),
     FromSql(usize),
@@ -379,7 +379,9 @@ impl fmt::Display for Error {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         match &self.0.kind {
             Kind::Io => fmt.write_str("error communicating with the server")?,
-            Kind::UnexpectedMessage => fmt.write_str("unexpected message from server")?,
+            Kind::UnexpectedMessage(msg) => {
+                write!(fmt, "unexpected message from server: {:?}", msg)?
+            }
             Kind::Tls => fmt.write_str("error performing TLS handshake")?,
             Kind::ToSql(idx) => write!(fmt, "error serializing parameter {}", idx)?,
             Kind::FromSql(idx) => write!(fmt, "error deserializing column {}", idx)?,
@@ -445,8 +447,8 @@ impl Error {
         Error::new(Kind::Closed, None)
     }
 
-    pub(crate) fn unexpected_message() -> Error {
-        Error::new(Kind::UnexpectedMessage, None)
+    pub(crate) fn unexpected_message(message: Message) -> Error {
+        Error::new(Kind::UnexpectedMessage(message), None)
     }
 
     #[allow(clippy::needless_pass_by_value)]