Start on std::futures rewrite

connect_raw works!

Author: sfackler

tokio-postgres/Cargo.toml

@@ -21,7 +21,7 @@ circle-ci = { repository = "sfackler/rust-postgres" }
 
 [features]
 default = ["runtime"]
-runtime = ["tokio-tcp", "tokio-timer", "tokio-uds", "tokio-threadpool", "lazy_static"]
+runtime = ["tokio/rt-full", "tokio/tcp", "tokio/uds", "tokio-threadpool", "lazy_static"]
 
 "with-bit-vec-0_5" = ["bit-vec-05"]
 "with-chrono-0_4" = ["chrono-04"]
@@ -34,17 +34,15 @@ with-serde_json-1 = ["serde-1", "serde_json-1"]
 antidote = "1.0"
 bytes = "0.4"
 fallible-iterator = "0.2"
+futures-preview = "0.3.0-alpha.17"
 log = "0.4"
 percent-encoding = "1.0"
 phf = "0.7.23"
 postgres-protocol = { version = "0.4.1", path = "../postgres-protocol" }
-tokio-codec = { git = "https://github.com/tokio-rs/tokio" }
-tokio-io = { git = "https://github.com/tokio-rs/tokio" }
+tokio = { git = "https://github.com/tokio-rs/tokio", default-features = false, features = ["io", "codec"] }
 
-tokio-tcp = { git = "https://github.com/tokio-rs/tokio", optional = true }
 tokio-threadpool = { git = "https://github.com/tokio-rs/tokio", optional = true }
 lazy_static = { version = "1.0", optional = true }
-tokio-timer = { git = "https://github.com/tokio-rs/tokio", optional = true }
 
 bit-vec-05 = { version = "0.5", package = "bit-vec", optional = true }
 chrono-04 = { version = "0.4", package = "chrono", optional = true }
@@ -54,9 +52,6 @@ serde-1 = { version = "1.0", package = "serde", optional = true }
 serde_json-1 = { version = "1.0", package = "serde_json", optional = true }
 uuid-07 = { version = "0.7", package = "uuid", optional = true }
 
-[target.'cfg(unix)'.dependencies]
-tokio-uds = { git = "https://github.com/tokio-rs/tokio", optional = true }
-
 [dev-dependencies]
 tokio = { git = "https://github.com/tokio-rs/tokio" }
 env_logger = "0.5"

tokio-postgres/src/client.rs

@@ -0,0 +1,22 @@
+use crate::connection::Request;
+use futures::channel::mpsc;
+
+pub struct Client {
+    sender: mpsc::UnboundedSender<Request>,
+    process_id: i32,
+    secret_key: i32,
+}
+
+impl Client {
+    pub(crate) fn new(
+        sender: mpsc::UnboundedSender<Request>,
+        process_id: i32,
+        secret_key: i32,
+    ) -> Client {
+        Client {
+            sender,
+            process_id,
+            secret_key,
+        }
+    }
+}

tokio-postgres/src/codec.rs

@@ -0,0 +1,99 @@
+use bytes::{Buf, BytesMut};
+use fallible_iterator::FallibleIterator;
+use postgres_protocol::message::backend;
+use postgres_protocol::message::frontend::CopyData;
+use std::io;
+use tokio::codec::{Decoder, Encoder};
+
+pub enum FrontendMessage {
+    Raw(Vec<u8>),
+    CopyData(CopyData<Box<dyn Buf + Send>>),
+}
+
+pub enum BackendMessage {
+    Normal {
+        messages: BackendMessages,
+        request_complete: bool,
+    },
+    Async(backend::Message),
+}
+
+pub struct BackendMessages(BytesMut);
+
+impl BackendMessages {
+    pub fn empty() -> BackendMessages {
+        BackendMessages(BytesMut::new())
+    }
+}
+
+impl FallibleIterator for BackendMessages {
+    type Item = backend::Message;
+    type Error = io::Error;
+
+    fn next(&mut self) -> io::Result<Option<backend::Message>> {
+        backend::Message::parse(&mut self.0)
+    }
+}
+
+pub struct PostgresCodec;
+
+impl Encoder for PostgresCodec {
+    type Item = FrontendMessage;
+    type Error = io::Error;
+
+    fn encode(&mut self, item: FrontendMessage, dst: &mut BytesMut) -> io::Result<()> {
+        match item {
+            FrontendMessage::Raw(buf) => dst.extend_from_slice(&buf),
+            FrontendMessage::CopyData(data) => data.write(dst),
+        }
+
+        Ok(())
+    }
+}
+
+impl Decoder for PostgresCodec {
+    type Item = BackendMessage;
+    type Error = io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<BackendMessage>, io::Error> {
+        let mut idx = 0;
+        let mut request_complete = false;
+
+        while let Some(header) = backend::Header::parse(&src[idx..])? {
+            let len = header.len() as usize + 1;
+            if src[idx..].len() < len {
+                break;
+            }
+
+            match header.tag() {
+                backend::NOTICE_RESPONSE_TAG
+                | backend::NOTIFICATION_RESPONSE_TAG
+                | backend::PARAMETER_STATUS_TAG => {
+                    if idx == 0 {
+                        let message = backend::Message::parse(src)?.unwrap();
+                        return Ok(Some(BackendMessage::Async(message)));
+                    } else {
+                        break;
+                    }
+                }
+                _ => {}
+            }
+
+            idx += len;
+
+            if header.tag() == backend::READY_FOR_QUERY_TAG {
+                request_complete = true;
+                break;
+            }
+        }
+
+        if idx == 0 {
+            Ok(None)
+        } else {
+            Ok(Some(BackendMessage::Normal {
+                messages: BackendMessages(src.split_to(idx)),
+                request_complete,
+            }))
+        }
+    }
+}