Update tokio-udp to use std-future (#1199)

Cargo.toml

@@ -20,6 +20,6 @@ members = [
   # "tokio-tls",
   # "tokio-trace",
   # "tokio-trace/tokio-trace-core",
-  # "tokio-udp",
+  "tokio-udp",
   # "tokio-uds",
 ]

azure-pipelines.yml

@@ -33,7 +33,7 @@ jobs:
 #       - tokio-signal
 #       - tokio-tcp
 #       - tokio-tls
-#       - tokio-udp
+      - tokio-udp
 #       - tokio-uds
 
 # Test crates that are NOT platform specific

tokio-macros/src/lib.rs

@@ -79,7 +79,7 @@ pub fn test(_attr: TokenStream, item: TokenStream) -> TokenStream {
         #(#attrs)*
         fn #name() #ret {
             let mut rt = tokio::runtime::current_thread::Runtime::new().unwrap();
-            rt.block_on_async(async { #body })
+            rt.block_on(async { #body })
         }
     };
 

tokio-udp/Cargo.toml

@@ -22,13 +22,13 @@ categories = ["asynchronous"]
 publish = false
 
 [dependencies]
-tokio-codec = { version = "0.2.0", path = "../tokio-codec" }
-tokio-io = { version = "0.2.0", path = "../tokio-io" }
+# tokio-codec = { version = "0.2.0", path = "../tokio-codec" }
+# tokio-io = { version = "0.2.0", path = "../tokio-io" }
 tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" }
-bytes = "0.4"
+# bytes = "0.4"
 mio = "0.6.14"
 log = "0.4"
-futures = "0.1.19"
 
 [dev-dependencies]
 env_logger = { version = "0.5", default-features = false }
+tokio = { version = "0.2.0", path = "../tokio", default-features = false, features = ["rt-full"] }

tokio-udp/src/lib.rs

@@ -10,23 +10,28 @@
 //!
 //! The main struct for UDP is the [`UdpSocket`], which represents a UDP socket.
 //! Reading and writing to it can be done using futures, which return the
-//! [`RecvDgram`] and [`SendDgram`] structs respectively.
-//!
-//! For convenience it's also possible to convert raw datagrams into higher-level
-//! frames.
-//!
-//! [`UdpSocket`]: struct.UdpSocket.html
-//! [`RecvDgram`]: struct.RecvDgram.html
-//! [`SendDgram`]: struct.SendDgram.html
-//! [`UdpFramed`]: struct.UdpFramed.html
-//! [`framed`]: struct.UdpSocket.html#method.framed
+//! [`Recv`], [`Send`], [`RecvFrom`] and [`SendTo`] structs respectively.
 
-mod frame;
-mod recv_dgram;
-mod send_dgram;
+macro_rules! ready {
+    ($e:expr) => {
+        match $e {
+            ::std::task::Poll::Ready(t) => t,
+            ::std::task::Poll::Pending => return ::std::task::Poll::Pending,
+        }
+    };
+}
+
+// mod frame;
+mod recv;
+mod recv_from;
+mod send;
+mod send_to;
 mod socket;
 
-pub use self::frame::UdpFramed;
-pub use self::recv_dgram::RecvDgram;
-pub use self::send_dgram::SendDgram;
+// pub use self::frame::UdpFramed;
+pub use self::recv::Recv;
+pub use self::recv_from::RecvFrom;
+pub use self::send::Send;
+pub use self::send_to::SendTo;
+
 pub use self::socket::UdpSocket;

tokio-udp/src/recv.rs

@@ -0,0 +1,30 @@
+use super::UdpSocket;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future that receives a datagram from the connected address.
+///
+/// This `struct` is created by [`recv`](super::UdpSocket::recv).
+#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+pub struct Recv<'a, 'b> {
+    socket: &'a mut UdpSocket,
+    buf: &'b mut [u8],
+}
+
+impl<'a, 'b> Recv<'a, 'b> {
+    pub(super) fn new(socket: &'a mut UdpSocket, buf: &'b mut [u8]) -> Self {
+        Self { socket, buf }
+    }
+}
+
+impl<'a, 'b> Future for Recv<'a, 'b> {
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Recv { socket, buf } = self.get_mut();
+        Pin::new(&mut **socket).poll_recv(cx, buf)
+    }
+}

tokio-udp/src/recv_from.rs

@@ -0,0 +1,31 @@
+use super::UdpSocket;
+use std::future::Future;
+use std::io;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future that receives a datagram.
+///
+/// This `struct` is created by [`recv_from`](super::UdpSocket::recv_from).
+#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+pub struct RecvFrom<'a, 'b> {
+    socket: &'a mut UdpSocket,
+    buf: &'b mut [u8],
+}
+
+impl<'a, 'b> RecvFrom<'a, 'b> {
+    pub(super) fn new(socket: &'a mut UdpSocket, buf: &'b mut [u8]) -> Self {
+        Self { socket, buf }
+    }
+}
+
+impl<'a, 'b> Future for RecvFrom<'a, 'b> {
+    type Output = io::Result<(usize, SocketAddr)>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let RecvFrom { socket, buf } = self.get_mut();
+        Pin::new(&mut **socket).poll_recv_from(cx, buf)
+    }
+}

tokio-udp/src/send.rs

@@ -0,0 +1,30 @@
+use super::UdpSocket;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future that sends a datagram to the connected address.
+///
+/// This `struct` is created by [`send`](super::UdpSocket::send).
+#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+pub struct Send<'a, 'b> {
+    socket: &'a mut UdpSocket,
+    buf: &'b [u8],
+}
+
+impl<'a, 'b> Send<'a, 'b> {
+    pub(super) fn new(socket: &'a mut UdpSocket, buf: &'b [u8]) -> Self {
+        Self { socket, buf }
+    }
+}
+
+impl<'a, 'b> Future for Send<'a, 'b> {
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Send { socket, buf } = self.get_mut();
+        Pin::new(&mut **socket).poll_send(cx, buf)
+    }
+}

tokio-udp/src/send_to.rs

@@ -0,0 +1,40 @@
+use super::UdpSocket;
+use std::future::Future;
+use std::io;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future that sends a datagram to a given address.
+///
+/// This `struct` is created by [`send_to`](super::UdpSocket::send_to).
+#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+pub struct SendTo<'a, 'b> {
+    socket: &'a mut UdpSocket,
+    buf: &'b [u8],
+    target: &'b SocketAddr,
+}
+
+impl<'a, 'b> SendTo<'a, 'b> {
+    pub(super) fn new(socket: &'a mut UdpSocket, buf: &'b [u8], target: &'b SocketAddr) -> Self {
+        Self {
+            socket,
+            buf,
+            target,
+        }
+    }
+}
+
+impl<'a, 'b> Future for SendTo<'a, 'b> {
+    type Output = io::Result<usize>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let SendTo {
+            socket,
+            buf,
+            target,
+        } = self.get_mut();
+        Pin::new(&mut **socket).poll_send_to(cx, buf, target)
+    }
+}