chore: remove tokio01::codec

src/internal_events/unix.rs

@@ -20,7 +20,7 @@ impl InternalEvent for UnixSocketConnectionEstablished<'_> {
 
 #[derive(Debug)]
 pub struct UnixSocketConnectionFailure<'a> {
-    pub error: std::io::Error,
+    pub error: tokio::io::Error,
     pub path: &'a std::path::Path,
 }
 

src/sinks/util/unix.rs

@@ -7,19 +7,18 @@ use crate::{
     sinks::{Healthcheck, RouterSink},
     topology::config::SinkContext,
 };
-use bytes::Bytes;
-use futures01::{
-    future, stream::iter_ok, try_ready, Async, AsyncSink, Future, Poll, Sink, StartSend,
-};
+use bytes05::Bytes;
+use futures::{compat::CompatSink, FutureExt, TryFutureExt};
+use futures01::{stream::iter_ok, try_ready, Async, AsyncSink, Future, Poll, Sink, StartSend};
 use serde::{Deserialize, Serialize};
 use snafu::Snafu;
-use std::io;
-use std::path::PathBuf;
-use std::time::{Duration, Instant};
-use tokio01::codec::{BytesCodec, FramedWrite};
-use tokio01::timer::Delay;
+use std::{path::PathBuf, time::Duration};
+use tokio::{
+    net::UnixStream,
+    time::{delay_for, Delay},
+};
 use tokio_retry::strategy::ExponentialBackoff;
-use tokio_uds::UnixStream;
+use tokio_util::codec::{BytesCodec, FramedWrite};
 use tracing::field;
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
@@ -41,27 +40,23 @@ impl UnixSinkConfig {
 
         let sink =
             Box::new(sink.with_flat_map(move |event| iter_ok(encode_event(event, &encoding))));
-        let healthcheck = unix_healthcheck(self.path.clone());
+        let healthcheck = healthcheck(self.path.clone()).boxed().compat();
 
-        Ok((sink, healthcheck))
+        Ok((sink, Box::new(healthcheck)))
     }
 }
 
 #[derive(Debug, Snafu)]
 enum HealthcheckError {
     #[snafu(display("Connect error: {}", source))]
-    ConnectError { source: std::io::Error },
+    ConnectError { source: tokio::io::Error },
 }
 
-fn unix_healthcheck(path: PathBuf) -> Healthcheck {
-    // Lazy to avoid immediately connecting
-    let check = future::lazy(move || {
-        UnixStream::connect(&path)
-            .map(|_| ())
-            .map_err(|source| HealthcheckError::ConnectError { source }.into())
-    });
-
-    Box::new(check)
+async fn healthcheck(path: PathBuf) -> crate::Result<()> {
+    match UnixStream::connect(&path).await {
+        Ok(_) => Ok(()),
+        Err(source) => Err(HealthcheckError::ConnectError { source }.into()),
+    }
 }
 
 pub struct UnixSink {
@@ -72,9 +67,9 @@ pub struct UnixSink {
 
 enum UnixSinkState {
     Disconnected,
-    Creating(Box<dyn Future<Item = UnixStream, Error = io::Error> + Send + 'static>),
-    Open(FramedWrite<UnixStream, BytesCodec>),
-    Backoff(Delay),
+    Creating(Box<dyn Future<Item = UnixStream, Error = tokio::io::Error> + Send + 'static>),
+    Open(CompatSink<FramedWrite<UnixStream, BytesCodec>, Bytes>),
+    Backoff(Box<dyn Future<Item = (), Error = ()> + Send>),
 }
 
 impl UnixSink {
@@ -94,13 +89,15 @@ impl UnixSink {
     }
 
     fn next_delay(&mut self) -> Delay {
-        Delay::new(Instant::now() + self.backoff.next().unwrap())
+        delay_for(self.backoff.next().unwrap())
     }
 
     /**
      * Polls for whether the underlying UnixStream is connected and ready to receive writes.
      **/
-    fn poll_connection(&mut self) -> Poll<&mut FramedWrite<UnixStream, BytesCodec>, ()> {
+    fn poll_connection(
+        &mut self,
+    ) -> Poll<&mut CompatSink<FramedWrite<UnixStream, BytesCodec>, Bytes>, ()> {
         loop {
             self.state = match self.state {
                 UnixSinkState::Open(ref mut stream) => {
@@ -115,27 +112,28 @@ impl UnixSink {
                             error,
                             path: &self.path
                         });
-                        UnixSinkState::Backoff(self.next_delay())
+                        let delay = self.next_delay();
+                        let delay = Box::new(async move { Ok(delay.await) }.boxed().compat());
+                        UnixSinkState::Backoff(delay)
                     }
                     Ok(Async::Ready(stream)) => {
                         emit!(UnixSocketConnectionEstablished { path: &self.path });
                         self.backoff = Self::fresh_backoff();
                         let out = FramedWrite::new(stream, BytesCodec::new());
-                        UnixSinkState::Open(out)
+                        UnixSinkState::Open(CompatSink::new(out))
                     }
                 },
                 UnixSinkState::Backoff(ref mut delay) => match delay.poll() {
                     Ok(Async::NotReady) => return Ok(Async::NotReady),
-                    // Err can only occur if the tokio runtime has been shutdown or if more than 2^63 timers have been created
-                    Err(err) => unreachable!(err),
+                    Err(_) => unreachable!(),
                     Ok(Async::Ready(())) => UnixSinkState::Disconnected,
                 },
                 UnixSinkState::Disconnected => {
                     debug!(
                         message = "connecting",
                         path = &field::display(self.path.to_str().unwrap())
                     );
-                    let connect_future = UnixStream::connect(&self.path);
+                    let connect_future = UnixStream::connect(self.path.clone()).boxed().compat();
                     UnixSinkState::Creating(Box::new(connect_future))
                 }
             }
@@ -144,7 +142,7 @@ impl UnixSink {
 }
 
 impl Sink for UnixSink {
-    type SinkItem = Bytes;
+    type SinkItem = bytes::Bytes;
     type SinkError = ();
 
     fn start_send(&mut self, line: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
@@ -154,20 +152,29 @@ impl Sink for UnixSink {
             Err(_) => {
                 unreachable!(); // poll_ready() should never return an error
             }
-            Ok(Async::Ready(connection)) => match connection.start_send(line) {
-                Err(error) => {
-                    emit!(UnixSocketError {
-                        error,
-                        path: &self.path
-                    });
-                    self.state = UnixSinkState::Disconnected;
-                    Ok(AsyncSink::Ready)
-                }
-                Ok(res) => {
-                    emit!(UnixSocketEventSent { byte_size });
-                    Ok(res)
+            Ok(Async::Ready(connection)) => {
+                let line = Bytes::copy_from_slice(&line);
+                match connection.start_send(line) {
+                    Err(error) => {
+                        emit!(UnixSocketError {
+                            error,
+                            path: &self.path
+                        });
+                        self.state = UnixSinkState::Disconnected;
+                        Ok(AsyncSink::Ready)
+                    }
+                    Ok(res) => {
+                        emit!(UnixSocketEventSent { byte_size });
+                        Ok(match res {
+                            AsyncSink::Ready => AsyncSink::Ready,
+                            AsyncSink::NotReady(bytes) => {
+                                let bytes = bytes::Bytes::from(&bytes[..]);
+                                AsyncSink::NotReady(bytes)
+                            }
+                        })
+                    }
                 }
-            },
+            }
         }
     }
 
@@ -208,14 +215,12 @@ mod tests {
 
     #[tokio::test]
     async fn unix_sink_healthcheck() {
-        let path = temp_uds_path("valid_uds");
-        let _listener = UnixListener::bind(&path).unwrap();
-        let healthcheck = unix_healthcheck(path);
-        assert!(healthcheck.wait().is_ok());
+        let good_path = temp_uds_path("valid_uds");
+        let _listener = UnixListener::bind(&good_path).unwrap();
+        assert!(healthcheck(good_path).await.is_ok());
 
         let bad_path = temp_uds_path("no_one_listening");
-        let bad_healthcheck = unix_healthcheck(bad_path);
-        assert!(bad_healthcheck.wait().is_err());
+        assert!(healthcheck(bad_path).await.is_err());
     }
 
     #[test]