change futures channel to tokio channel (#28)

Cargo.toml

@@ -21,14 +21,14 @@ rustls-tls = ["tokio-rustls", "rustls-pki-types"]
 
 [dependencies]
 bytes = "1.5.0"
-futures-channel = "0.3.30"
 futures-util = { version = "0.3.30", features = ["sink"] }
 hostname = "0.3.1"
 rustls-pki-types = { version = "1.1.0", optional = true }
 serde_json = "1.0.108"
 thiserror = "1.0.52"
-tokio = { version = "1.35.1", features = ["io-util", "net", "time"] }
+tokio = { version = "1.35.1", features = ["io-util", "net", "time", "sync"] }
 tokio-rustls = { version = "0.25.0", optional = true }
+tokio-stream = "0.1.15"
 tokio-util = { version = "0.7.10", features = ["codec", "net"] }
 tracing-core = "0.1.32"
 tracing-futures = "0.2.5"

src/connection/mod.rs

@@ -4,8 +4,8 @@ mod udp;
 use std::{io, net::SocketAddr};
 
 use bytes::Bytes;
-use futures_channel::mpsc;
 use tokio::net::{lookup_host, ToSocketAddrs};
+use tokio_stream::wrappers::ReceiverStream;
 use tracing_core::subscriber::NoSubscriber;
 use tracing_futures::WithSubscriber;
 
@@ -26,7 +26,7 @@ pub struct ConnectionErrors(pub Vec<(SocketAddr, io::Error)>);
 #[must_use]
 pub struct ConnectionHandle<A, Conn> {
     pub(crate) addr: A,
-    pub(crate) receiver: mpsc::Receiver<Bytes>,
+    pub(crate) receiver: ReceiverStream<Bytes>,
     pub(crate) conn: Conn,
 }
 

src/lib.rs

@@ -59,9 +59,10 @@ mod visitor;
 use std::{borrow::Cow, collections::HashMap, fmt::Display};
 
 use bytes::Bytes;
-use futures_channel::mpsc;
 use serde_json::{map::Map, Value};
 use tokio::net::ToSocketAddrs;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
 use tracing_core::{
     dispatcher::SetGlobalDefaultError,
     span::{Attributes, Id, Record},
@@ -248,7 +249,7 @@ impl Builder {
         let (sender, receiver) = mpsc::channel::<Bytes>(buffer);
         let handle = ConnectionHandle {
             addr,
-            receiver,
+            receiver: ReceiverStream::new(receiver),
             conn,
         };
         let logger = Logger {