Switch copy_in to use Buf

Author: sfackler

tokio-postgres/src/lib.rs

@@ -16,7 +16,7 @@ extern crate log;
 #[macro_use]
 extern crate state_machine_future;
 
-use bytes::Bytes;
+use bytes::{Bytes, IntoBuf};
 use futures::{Async, Future, Poll, Stream};
 use postgres_shared::rows::RowIndex;
 use std::error::Error as StdError;
@@ -90,7 +90,8 @@ impl Client {
     pub fn copy_in<S>(&mut self, statement: &Statement, params: &[&ToSql], stream: S) -> CopyIn<S>
     where
         S: Stream,
-        S::Item: AsRef<[u8]>,
+        S::Item: IntoBuf,
+        <S::Item as IntoBuf>::Buf: Send,
         // FIXME error type?
         S::Error: Into<Box<StdError + Sync + Send>>,
     {
@@ -283,13 +284,15 @@ pub struct Portal(proto::Portal);
 pub struct CopyIn<S>(proto::CopyInFuture<S>)
 where
     S: Stream,
-    S::Item: AsRef<[u8]>,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
     S::Error: Into<Box<StdError + Sync + Send>>;
 
 impl<S> Future for CopyIn<S>
 where
     S: Stream,
-    S::Item: AsRef<[u8]>,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
     S::Error: Into<Box<StdError + Sync + Send>>,
 {
     type Item = u64;

tokio-postgres/src/proto/client.rs

@@ -1,4 +1,5 @@
 use antidote::Mutex;
+use bytes::IntoBuf;
 use futures::sync::mpsc;
 use futures::{AsyncSink, Sink, Stream};
 use postgres_protocol;
@@ -163,7 +164,8 @@ impl Client {
     pub fn copy_in<S>(&self, statement: &Statement, params: &[&ToSql], stream: S) -> CopyInFuture<S>
     where
         S: Stream,
-        S::Item: AsRef<[u8]>,
+        S::Item: IntoBuf,
+        <S::Item as IntoBuf>::Buf: Send,
         S::Error: Into<Box<StdError + Sync + Send>>,
     {
         let (mut sender, receiver) = mpsc::channel(0);

tokio-postgres/src/proto/copy_in.rs

@@ -1,3 +1,4 @@
+use bytes::{Buf, IntoBuf};
 use futures::sink;
 use futures::sync::mpsc;
 use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
@@ -63,7 +64,8 @@ impl Stream for CopyInReceiver {
 pub enum CopyIn<S>
 where
     S: Stream,
-    S::Item: AsRef<[u8]>,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
     S::Error: Into<Box<StdError + Sync + Send>>,
 {
     #[state_machine_future(start, transitions(ReadCopyInResponse))]
@@ -103,7 +105,8 @@ where
 impl<S> PollCopyIn<S> for CopyIn<S>
 where
     S: Stream,
-    S::Item: AsRef<[u8]>,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
     S::Error: Into<Box<StdError + Sync + Send>>,
 {
     fn poll_start<'a>(state: &'a mut RentToOwn<'a, Start<S>>) -> Poll<AfterStart<S>, Error> {
@@ -151,7 +154,9 @@ where
                 None => match try_ready!(state.stream.poll().map_err(Error::copy_in_stream)) {
                     Some(data) => {
                         let mut buf = vec![];
-                        frontend::copy_data(data.as_ref(), &mut buf).map_err(Error::encode)?;
+                        // FIXME avoid collect
+                        frontend::copy_data(&data.into_buf().collect::<Vec<_>>(), &mut buf)
+                            .map_err(Error::encode)?;
                         CopyMessage::Data(buf)
                     }
                     None => {
@@ -213,7 +218,8 @@ where
 impl<S> CopyInFuture<S>
 where
     S: Stream,
-    S::Item: AsRef<[u8]>,
+    S::Item: IntoBuf,
+    <S::Item as IntoBuf>::Buf: Send,
     S::Error: Into<Box<StdError + Sync + Send>>,
 {
     pub fn new(