Move futures/streams to a submodule

Author: sfackler

postgres/src/copy_out_reader.rs

@@ -2,10 +2,11 @@ use bytes::{Buf, Bytes};
 use futures::stream::{self, Stream};
 use std::io::{self, BufRead, Cursor, Read};
 use std::marker::PhantomData;
+use tokio_postgres::impls;
 use tokio_postgres::Error;
 
 pub struct CopyOutReader<'a> {
-    it: stream::Wait<tokio_postgres::CopyOut>,
+    it: stream::Wait<impls::CopyOut>,
     cur: Cursor<Bytes>,
     _p: PhantomData<&'a mut ()>,
 }
@@ -17,7 +18,7 @@ impl<'a> Drop for CopyOutReader<'a> {
 
 impl<'a> CopyOutReader<'a> {
     #[allow(clippy::new_ret_no_self)]
-    pub(crate) fn new(stream: tokio_postgres::CopyOut) -> Result<CopyOutReader<'a>, Error> {
+    pub(crate) fn new(stream: impls::CopyOut) -> Result<CopyOutReader<'a>, Error> {
         let mut it = stream.wait();
         let cur = match it.next() {
             Some(Ok(cur)) => cur,

postgres/src/query.rs

@@ -1,10 +1,11 @@
 use fallible_iterator::FallibleIterator;
 use futures::stream::{self, Stream};
 use std::marker::PhantomData;
+use tokio_postgres::impls;
 use tokio_postgres::{Error, Row};
 
 pub struct Query<'a> {
-    it: stream::Wait<tokio_postgres::Query>,
+    it: stream::Wait<impls::Query>,
     _p: PhantomData<&'a mut ()>,
 }
 
@@ -14,7 +15,7 @@ impl<'a> Drop for Query<'a> {
 }
 
 impl<'a> Query<'a> {
-    pub(crate) fn new(stream: tokio_postgres::Query) -> Query<'a> {
+    pub(crate) fn new(stream: impls::Query) -> Query<'a> {
         Query {
             it: stream.wait(),
             _p: PhantomData,

postgres/src/query_portal.rs

@@ -1,10 +1,11 @@
 use fallible_iterator::FallibleIterator;
 use futures::stream::{self, Stream};
 use std::marker::PhantomData;
+use tokio_postgres::impls;
 use tokio_postgres::{Error, Row};
 
 pub struct QueryPortal<'a> {
-    it: stream::Wait<tokio_postgres::QueryPortal>,
+    it: stream::Wait<impls::QueryPortal>,
     _p: PhantomData<&'a mut ()>,
 }
 
@@ -14,7 +15,7 @@ impl<'a> Drop for QueryPortal<'a> {
 }
 
 impl<'a> QueryPortal<'a> {
-    pub(crate) fn new(stream: tokio_postgres::QueryPortal) -> QueryPortal<'a> {
+    pub(crate) fn new(stream: impls::QueryPortal) -> QueryPortal<'a> {
         QueryPortal {
             it: stream.wait(),
             _p: PhantomData,

tokio-postgres/src/config.rs

@@ -15,12 +15,15 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio_io::{AsyncRead, AsyncWrite};
 
+#[cfg(feature = "runtime")]
+use crate::impls::Connect;
+use crate::impls::ConnectRaw;
 #[cfg(feature = "runtime")]
 use crate::proto::ConnectFuture;
 use crate::proto::ConnectRawFuture;
+use crate::{Error, TlsConnect};
 #[cfg(feature = "runtime")]
-use crate::{Connect, MakeTlsConnect, Socket};
-use crate::{ConnectRaw, Error, TlsConnect};
+use crate::{MakeTlsConnect, Socket};
 
 /// Properties required of a session.
 #[cfg(feature = "runtime")]

tokio-postgres/src/impls.rs

@@ -0,0 +1,178 @@
+use bytes::{Bytes, IntoBuf};
+use futures::{try_ready, Async, Future, Poll, Stream};
+use std::error;
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use crate::proto;
+use crate::{Client, Connection, Error, Portal, Row, Statement, TlsConnect};
+#[cfg(feature = "runtime")]
+use crate::{MakeTlsConnect, Socket};
+
+#[must_use = "futures do nothing unless polled"]
+pub struct CancelQueryRaw<S, T>(pub(crate) proto::CancelQueryRawFuture<S, T>)
+where
+    S: AsyncRead + AsyncWrite,
+    T: TlsConnect<S>;
+
+impl<S, T> Future for CancelQueryRaw<S, T>
+where
+    S: AsyncRead + AsyncWrite,
+    T: TlsConnect<S>,
+{
+    type Item = ();
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<(), Error> {
+        self.0.poll()
+    }
+}
+
+#[cfg(feature = "runtime")]
+#[must_use = "futures do nothing unless polled"]
+pub struct CancelQuery<T>(pub(crate) proto::CancelQueryFuture<T>)
+where
+    T: MakeTlsConnect<Socket>;
+
+#[cfg(feature = "runtime")]
+impl<T> Future for CancelQuery<T>
+where
+    T: MakeTlsConnect<Socket>,
+{
+    type Item = ();
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<(), Error> {
+        self.0.poll()
+    }
+}
+
+#[must_use = "futures do nothing unless polled"]
+pub struct ConnectRaw<S, T>(pub(crate) proto::ConnectRawFuture<S, T>)
+where
+    S: AsyncRead + AsyncWrite,
+    T: TlsConnect<S>;
+
+impl<S, T> Future for ConnectRaw<S, T>
+where
+    S: AsyncRead + AsyncWrite,
+    T: TlsConnect<S>,
+{
+    type Item = (Client, Connection<S, T::Stream>);
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<(Client, Connection<S, T::Stream>), Error> {
+        let (client, connection) = try_ready!(self.0.poll());
+
+        Ok(Async::Ready((Client(client), Connection(connection))))
+    }
+}
+
+#[cfg(feature = "runtime")]
+#[must_use = "futures do nothing unless polled"]
+pub struct Connect<T>(pub(crate) proto::ConnectFuture<T>)
+where
+    T: MakeTlsConnect<Socket>;
+
+#[cfg(feature = "runtime")]
+impl<T> Future for Connect<T>
+where
+    T: MakeTlsConnect<Socket>,
+{
+    type Item = (Client, Connection<Socket, T::Stream>);
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<(Client, Connection<Socket, T::Stream>), Error> {
+        let (client, connection) = try_ready!(self.0.poll());
+
+        Ok(Async::Ready((Client(client), Connection(connection))))
+    }
+}
+
+#[must_use = "futures do nothing unless polled"]
+pub struct Prepare(pub(crate) proto::PrepareFuture);
+
+impl Future for Prepare {
+    type Item = Statement;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Statement, Error> {
+        let statement = try_ready!(self.0.poll());
+
+        Ok(Async::Ready(Statement(statement)))
+    }
+}
+
+#[must_use = "streams do nothing unless polled"]
+pub struct Query(pub(crate) proto::QueryStream<proto::Statement>);
+
+impl Stream for Query {
+    type Item = Row;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Row>, Error> {
+        self.0.poll()
+    }
+}
+
+#[must_use = "futures do nothing unless polled"]
+pub struct Bind(pub(crate) proto::BindFuture);
+
+impl Future for Bind {
+    type Item = Portal;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Portal, Error> {
+        match self.0.poll() {
+            Ok(Async::Ready(portal)) => Ok(Async::Ready(Portal(portal))),
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Err(e) => Err(e),
+        }
+    }
+}
+
+#[must_use = "streams do nothing unless polled"]
+pub struct QueryPortal(pub(crate) proto::QueryStream<proto::Portal>);
+
+impl Stream for QueryPortal {
+    type Item = Row;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Row>, Error> {
+        self.0.poll()
+    }
+}
+
+#[must_use = "futures do nothing unless polled"]
+pub struct CopyIn<S>(pub(crate) proto::CopyInFuture<S>)
+where
+    S: Stream,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
+    S::Error: Into<Box<dyn error::Error + Sync + Send>>;
+
+impl<S> Future for CopyIn<S>
+where
+    S: Stream,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
+    S::Error: Into<Box<dyn error::Error + Sync + Send>>,
+{
+    type Item = u64;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<u64, Error> {
+        self.0.poll()
+    }
+}
+
+#[must_use = "streams do nothing unless polled"]
+pub struct CopyOut(pub(crate) proto::CopyOutStream);
+
+impl Stream for CopyOut {
+    type Item = Bytes;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Bytes>, Error> {
+        self.0.poll()
+    }
+}