Make streams !Unpin

Hedging against the future if we end up using an intrusive list
implementation

Author: sfackler

tokio-postgres/Cargo.toml

@@ -38,6 +38,7 @@ log = "0.4"
 parking_lot = "0.9"
 percent-encoding = "1.0"
 pin-utils = "=0.1.0-alpha.4"
+pin-project = "0.4"
 phf = "0.7.23"
 postgres-protocol = { version = "0.4.1", path = "../postgres-protocol" }
 postgres-types = { version = "0.1.0", path = "../postgres-types" }

tokio-postgres/src/lib.rs

@@ -114,6 +114,8 @@ pub use crate::row::{Row, SimpleQueryRow};
 #[cfg(feature = "runtime")]
 pub use crate::socket::Socket;
 pub use crate::statement::{Column, Statement};
+pub use crate::query::RowStream;
+pub use crate::simple_query::SimpleQueryStream;
 #[cfg(feature = "runtime")]
 use crate::tls::MakeTlsConnect;
 pub use crate::tls::NoTls;

tokio-postgres/src/query.rs

@@ -1,4 +1,5 @@
 use crate::client::{InnerClient, Responses};
+use pin_project::pin_project;
 use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
 use crate::types::{IsNull, ToSql};
@@ -9,6 +10,7 @@ use postgres_protocol::message::backend::Message;
 use postgres_protocol::message::frontend;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use std::marker::PhantomPinned;
 
 pub async fn query<'a, I>(
     client: &InnerClient,
@@ -24,6 +26,7 @@ where
     Ok(RowStream {
         statement,
         responses,
+        _p: PhantomPinned,
     })
 }
 
@@ -43,6 +46,7 @@ pub async fn query_portal(
     Ok(RowStream {
         statement: portal.statement().clone(),
         responses,
+        _p: PhantomPinned,
     })
 }
 
@@ -145,18 +149,23 @@ where
     }
 }
 
+/// A stream of table rows.
+#[pin_project]
 pub struct RowStream {
     statement: Statement,
     responses: Responses,
+    #[pin]
+    _p: PhantomPinned,
 }
 
 impl Stream for RowStream {
     type Item = Result<Row, Error>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match ready!(self.responses.poll_next(cx)?) {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+        match ready!(this.responses.poll_next(cx)?) {
             Message::DataRow(body) => {
-                Poll::Ready(Some(Ok(Row::new(self.statement.clone(), body)?)))
+                Poll::Ready(Some(Ok(Row::new(this.statement.clone(), body)?)))
             }
             Message::EmptyQueryResponse
             | Message::CommandComplete(_)

tokio-postgres/src/simple_query.rs

@@ -10,6 +10,8 @@ use postgres_protocol::message::frontend;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
+use std::marker::PhantomPinned;
+use pin_project::pin_project;
 
 pub async fn simple_query(client: &InnerClient, query: &str) -> Result<SimpleQueryStream, Error> {
     let buf = encode(client, query)?;
@@ -18,6 +20,7 @@ pub async fn simple_query(client: &InnerClient, query: &str) -> Result<SimpleQue
     Ok(SimpleQueryStream {
         responses,
         columns: None,
+        _p: PhantomPinned,
     })
 }
 
@@ -44,17 +47,22 @@ fn encode(client: &InnerClient, query: &str) -> Result<Bytes, Error> {
     })
 }
 
+/// A stream of simple query results.
+#[pin_project]
 pub struct SimpleQueryStream {
     responses: Responses,
     columns: Option<Arc<[String]>>,
+    #[pin]
+    _p: PhantomPinned,
 }
 
 impl Stream for SimpleQueryStream {
     type Item = Result<SimpleQueryMessage, Error>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
         loop {
-            match ready!(self.responses.poll_next(cx)?) {
+            match ready!(this.responses.poll_next(cx)?) {
                 Message::CommandComplete(body) => {
                     let rows = body
                         .tag()
@@ -76,10 +84,10 @@ impl Stream for SimpleQueryStream {
                         .collect::<Vec<_>>()
                         .map_err(Error::parse)?
                         .into();
-                    self.columns = Some(columns);
+                    *this.columns = Some(columns);
                 }
                 Message::DataRow(body) => {
-                    let row = match &self.columns {
+                    let row = match &this.columns {
                         Some(columns) => SimpleQueryRow::new(columns.clone(), body)?,
                         None => return Poll::Ready(Some(Err(Error::unexpected_message()))),
                     };