Remove log and env_log depndencies

Author: cowlicks

Cargo.toml

@@ -51,15 +51,13 @@ path = "../core"
 async-std = { version = "1.12.0", features = ["attributes", "unstable"] }
 async-compat = "0.2.1"
 tokio = { version = "1.27.0", features = ["macros", "net", "process", "rt", "rt-multi-thread", "sync", "time"] }
-env_logger = "0.7.1"
 anyhow = "1.0.28"
 instant = "0.1"
 criterion = { version = "0.4", features = ["async_std"] }
 pretty-bytes = "0.2.2"
 duplexify = "1.1.0"
 sluice = "0.5.4"
 futures = "0.3.13"
-log = "0.4"
 tracing-subscriber = { version = "0.3.19", features = ["env-filter", "fmt"] }
 tracing-tree = "0.4.0"
 tokio-util = { version = "0.7.14", features = ["compat"] }

benches/pipe.rs

@@ -1,21 +1,23 @@
+#[path = "../src/test_utils.rs"]
+mod test_utils;
 use async_std::task;
 use criterion::{criterion_group, criterion_main, Criterion, Throughput};
 use futures::{
     io::{AsyncRead, AsyncWrite},
     stream::StreamExt,
 };
 use hypercore_protocol::{schema::*, Channel, Duplex, Event, Message, Protocol, ProtocolBuilder};
-use log::*;
 use pretty_bytes::converter::convert as pretty_bytes;
 use sluice::pipe::pipe;
 use std::{io::Result, time::Instant};
+use tracing::{debug, error};
 
 const COUNT: u64 = 1000;
 const SIZE: u64 = 100;
 const CONNS: u64 = 10;
 
 fn bench_throughput(c: &mut Criterion) {
-    env_logger::from_env(env_logger::Env::default().default_filter_or("error")).init();
+    test_utils::log();
     let mut group = c.benchmark_group("pipe");
     group.sample_size(10);
     group.throughput(Throughput::Bytes(SIZE * COUNT * CONNS));

benches/throughput.rs

@@ -1,3 +1,5 @@
+#[path = "../src/test_utils.rs"]
+mod test_utils;
 use async_std::{
     net::{Shutdown, TcpListener, TcpStream},
     task,
@@ -9,16 +11,16 @@ use futures::{
     stream::{FuturesUnordered, StreamExt},
 };
 use hypercore_protocol::{schema::*, Channel, Event, Message, ProtocolBuilder};
-use log::*;
 use std::time::Instant;
+use tracing::{debug, info, trace};
 
 const PORT: usize = 11011;
 const SIZE: u64 = 1000;
 const COUNT: u64 = 200;
 const CLIENTS: usize = 1;
 
 fn bench_throughput(c: &mut Criterion) {
-    env_logger::from_env(env_logger::Env::default().default_filter_or("error")).init();
+    test_utils::log();
     let address = format!("localhost:{}", PORT);
 
     let mut group = c.benchmark_group("throughput");
@@ -67,7 +69,7 @@ criterion_main!(server_benches);
 
 async fn start_server(address: &str) -> futures::channel::oneshot::Sender<()> {
     let listener = TcpListener::bind(&address).await.unwrap();
-    log::info!("listening on {}", listener.local_addr().unwrap());
+    info!("listening on {}", listener.local_addr().unwrap());
     let (kill_tx, mut kill_rx) = futures::channel::oneshot::channel();
     task::spawn(async move {
         let mut incoming = listener.incoming();

examples/replication.rs

@@ -1,3 +1,5 @@
+#[path = "../src/test_utils.rs"]
+mod test_utils;
 use anyhow::Result;
 use async_std::{
     net::{TcpListener, TcpStream},
@@ -10,13 +12,13 @@ use hypercore::{
     Hypercore, HypercoreBuilder, PartialKeypair, RequestBlock, RequestUpgrade, Storage,
     VerifyingKey,
 };
-use std::{collections::HashMap, convert::TryInto, env, fmt::Debug, sync::OnceLock};
-use tracing::{error, info};
+use std::{collections::HashMap, convert::TryInto, env, fmt::Debug};
+use tracing::{error, info, instrument};
 
 use hypercore_protocol::{discovery_key, schema::*, Channel, Event, Message, ProtocolBuilder};
 
 fn main() {
-    log();
+    test_utils::log();
     if env::args().count() < 3 {
         usage();
     }
@@ -62,12 +64,11 @@ fn main() {
         hypercore_store.add(hypercore_wrapper);
         let hypercore_store = Arc::new(hypercore_store);
 
-        let result = match mode.as_ref() {
+        let _ = match mode.as_ref() {
             "server" => tcp_server(address, onconnection, hypercore_store).await,
             "client" => tcp_client(address, onconnection, hypercore_store).await,
             _ => panic!("{:?}", usage()),
         };
-        log_if_error(&result);
     });
 }
 
@@ -81,6 +82,7 @@ fn usage() {
 // or once when connected (if client).
 // Unfortunately, everything that touches the hypercore_store or a hypercore has to be generic
 // at the moment.
+#[instrument(skip_all, ret)]
 async fn onconnection(
     stream: TcpStream,
     is_initiator: bool,
@@ -123,17 +125,17 @@ struct HypercoreStore {
     hypercores: HashMap<String, Arc<HypercoreWrapper>>,
 }
 impl HypercoreStore {
-    pub fn new() -> Self {
+    fn new() -> Self {
         let hypercores = HashMap::new();
         Self { hypercores }
     }
 
-    pub fn add(&mut self, hypercore: HypercoreWrapper) {
+    fn add(&mut self, hypercore: HypercoreWrapper) {
         let hdkey = hex::encode(hypercore.discovery_key);
         self.hypercores.insert(hdkey, Arc::new(hypercore));
     }
 
-    pub fn get(&self, discovery_key: &[u8; 32]) -> Option<&Arc<HypercoreWrapper>> {
+    fn get(&self, discovery_key: &[u8; 32]) -> Option<&Arc<HypercoreWrapper>> {
         let hdkey = hex::encode(discovery_key);
         self.hypercores.get(&hdkey)
     }
@@ -148,7 +150,7 @@ struct HypercoreWrapper {
 }
 
 impl HypercoreWrapper {
-    pub fn from_memory_hypercore(hypercore: Hypercore) -> Self {
+    fn from_memory_hypercore(hypercore: Hypercore) -> Self {
         let key = hypercore.key_pair().public.to_bytes();
         HypercoreWrapper {
             key,
@@ -157,11 +159,11 @@ impl HypercoreWrapper {
         }
     }
 
-    pub fn key(&self) -> &[u8; 32] {
+    fn key(&self) -> &[u8; 32] {
         &self.key
     }
 
-    pub fn onpeer(&self, mut channel: Channel) {
+    fn onpeer(&self, mut channel: Channel) {
         let mut peer_state = PeerState::default();
         let mut hypercore = self.hypercore.clone();
         task::spawn(async move {
@@ -415,32 +417,9 @@ async fn onmessage(
     Ok(())
 }
 
-#[allow(unused)]
-pub fn log() {
-    use tracing_subscriber::{fmt::format::FmtSpan, EnvFilter};
-    static START_LOGS: OnceLock<()> = OnceLock::new();
-    START_LOGS.get_or_init(|| {
-        tracing_subscriber::fmt()
-            .with_target(true)
-            .with_line_number(true)
-            // print when instrumented funtion enters
-            .with_span_events(FmtSpan::ENTER | FmtSpan::EXIT)
-            .with_file(true)
-            .with_env_filter(EnvFilter::from_default_env()) // Reads `RUST_LOG` environment variable
-            .without_time()
-            .init();
-    });
-}
-
-/// Log a result if it's an error.
-pub fn log_if_error(result: &Result<()>) {
-    if let Err(err) = result.as_ref() {
-        log::error!("error: {}", err);
-    }
-}
-
 /// A simple async TCP server that calls an async function for each incoming connection.
-pub async fn tcp_server<F, C>(
+#[instrument(skip_all, ret)]
+async fn tcp_server<F, C>(
     address: String,
     onconnection: impl Fn(TcpStream, bool, C) -> F + Send + Sync + Copy + 'static,
     context: C,
@@ -450,22 +429,22 @@ where
     C: Clone + Send + 'static,
 {
     let listener = TcpListener::bind(&address).await?;
-    log::info!("listening on {}", listener.local_addr()?);
+    tracing::info!("listening on {}", listener.local_addr()?);
     let mut incoming = listener.incoming();
     while let Some(Ok(stream)) = incoming.next().await {
         let context = context.clone();
         let peer_addr = stream.peer_addr().unwrap();
-        log::info!("new connection from {}", peer_addr);
+        tracing::info!("new connection from {}", peer_addr);
         task::spawn(async move {
-            let result = onconnection(stream, false, context).await;
-            log_if_error(&result);
-            log::info!("connection closed from {}", peer_addr);
+            let _ = onconnection(stream, false, context).await;
+            tracing::info!("connection closed from {}", peer_addr);
         });
     }
     Ok(())
 }
 
 /// A simple async TCP client that calls an async function when connected.
+#[instrument(skip_all, ret)]
 pub async fn tcp_client<F, C>(
     address: String,
     onconnection: impl Fn(TcpStream, bool, C) -> F + Send + Sync + Copy + 'static,
@@ -475,8 +454,8 @@ where
     F: Future<Output = Result<()>> + Send,
     C: Clone + Send + 'static,
 {
-    log::info!("attempting connection to {address}");
+    tracing::info!("attempting connection to {address}");
     let stream = TcpStream::connect(&address).await?;
-    log::info!("connected to {address}");
+    tracing::info!("connected to {address}");
     onconnection(stream, true, context).await
 }

src/test_utils.rs

@@ -1,13 +1,13 @@
+#![allow(dead_code)]
 use std::{
     io::{self, ErrorKind},
     pin::Pin,
     task::{Context, Poll},
 };
 
-//use async_channel::{unbounded, Receiver, io::Error, Sender};
 use futures::{
     channel::mpsc::{unbounded, UnboundedReceiver as Receiver, UnboundedSender as Sender},
-    Sink, SinkExt, Stream, StreamExt,
+    Sink, Stream, StreamExt,
 };
 
 #[derive(Debug)]
@@ -99,6 +99,7 @@ pub(crate) fn log() {
 
 #[tokio::test]
 async fn way_one() {
+    use futures::SinkExt;
     let mut a = Io::default();
     let _ = a.send(b"hello".into()).await;
     let Some(res) = a.next().await else { panic!() };
@@ -107,6 +108,7 @@ async fn way_one() {
 
 #[tokio::test]
 async fn split() {
+    use futures::SinkExt;
     let (mut left, mut right) = (TwoWay::default()).split_sides();
 
     left.send(b"hello".to_vec()).await.unwrap();

tests/js_interop.rs

@@ -1,3 +1,5 @@
+#[path = "../src/test_utils.rs"]
+mod test_utils;
 use _util::wait_for_localhost_port;
 use anyhow::Result;
 use futures::Future;