Migrate from log to tracing

Cargo.toml

@@ -10,10 +10,11 @@ edition = "2021"
 anyhow = "1.0.53"
 clap = { version = "4.3.0", features = ["derive", "env"] }
 directories = "4.0.1"
-env_logger = "0.10.0"
-log = "0.4.14"
+pin-project-lite = "0.2.13"
 serde = { version = "1.0.186" }
 serde_derive = { version = "1.0.186" }
 serde_json = "1.0.78"
 tokio = { version = "1.15.0", features = ["fs", "io-std", "io-util", "macros", "net", "parking_lot", "process", "rt-multi-thread", "sync", "time"] }
 toml = "0.5.8"
+tracing = "0.1.39"
+tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }

rustfmt.toml

@@ -0,0 +1,2 @@
+group_imports = "StdExternalCrate"
+imports_granularity = "Module"

src/client.rs

@@ -1,9 +1,52 @@
-use anyhow::{Context, Result};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use anyhow::{Context as _, Result};
+use pin_project_lite::pin_project;
 use ra_multiplex::config::Config;
 use ra_multiplex::proto;
-use tokio::io::AsyncWriteExt;
+use tokio::io::{self, AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio::net::TcpStream;
-use tokio::{io, task};
+
+pin_project! {
+    struct Stdio {
+        #[pin]
+        stdin: io::Stdin,
+        #[pin]
+        stdout: io::Stdout,
+    }
+}
+
+fn stdio() -> Stdio {
+    Stdio {
+        stdin: io::stdin(),
+        stdout: io::stdout(),
+    }
+}
+
+impl AsyncRead for Stdio {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut io::ReadBuf,
+    ) -> Poll<io::Result<()>> {
+        self.project().stdin.poll_read(cx, buf)
+    }
+}
+
+impl AsyncWrite for Stdio {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<io::Result<usize>> {
+        self.project().stdout.poll_write(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
+        self.project().stdout.poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
+        self.project().stdout.poll_shutdown(cx)
+    }
+}
 
 pub async fn main(server_path: String, server_args: Vec<String>) -> Result<()> {
     let config = Config::load_or_default().await;
@@ -12,31 +55,18 @@ pub async fn main(server_path: String, server_args: Vec<String>) -> Result<()> {
     let mut proto_init = serde_json::to_vec(&proto_init).context("sending proto init")?;
     proto_init.push(b'\0');
 
-    let stream = TcpStream::connect(config.connect)
+    let mut stream = TcpStream::connect(config.connect)
         .await
         .context("connect")?;
-    let (mut read_stream, mut write_stream) = stream.into_split();
 
-    write_stream
+    stream
         .write_all(&proto_init)
         .await
         .context("sending proto init")?;
     drop(proto_init);
 
-    let t1 = task::spawn(async move {
-        io::copy(&mut read_stream, &mut io::stdout())
-            .await
-            .context("io error")
-    });
-    let t2 = task::spawn(async move {
-        io::copy(&mut io::stdin(), &mut write_stream)
-            .await
-            .context("io error")
-    });
-    tokio::select! {
-        res = t1 => res,
-        res = t2 => res,
-    }
-    .context("join")??;
+    io::copy_bidirectional(&mut stream, &mut stdio())
+        .await
+        .context("io error")?;
     Ok(())
 }

src/config.rs

@@ -1,11 +1,13 @@
+use std::net::{IpAddr, Ipv4Addr};
+
 use anyhow::Context;
 use directories::ProjectDirs;
 use serde::de::{Error, Unexpected};
 use serde::{Deserialize, Deserializer};
 use serde_derive::{Deserialize, Serialize};
-use std::net::{IpAddr, Ipv4Addr};
 use tokio::fs;
 use tokio::sync::OnceCell;
+use tracing::info;
 
 mod default {
     use super::*;
@@ -147,10 +149,20 @@ impl Config {
 
     /// panics if called multiple times
     fn init_logger(&self) {
-        env_logger::Builder::from_env(env_logger::Env::new().default_filter_or(&self.log_filters))
-            .format_timestamp(None)
-            .format_module_path(false)
-            .format_target(false)
+        use tracing_subscriber::prelude::*;
+        use tracing_subscriber::EnvFilter;
+
+        let format = tracing_subscriber::fmt::layer()
+            .without_time()
+            .with_target(false);
+
+        let filter = EnvFilter::try_from_default_env()
+            .or_else(|_| EnvFilter::try_new(&self.log_filters))
+            .unwrap_or_else(|_| EnvFilter::new("info"));
+
+        tracing_subscriber::registry()
+            .with(filter)
+            .with(format)
             .init();
     }
 
@@ -168,9 +180,9 @@ impl Config {
                 };
                 let global_config = Box::leak(Box::new(config));
                 global_config.init_logger();
-                if let Some(load_err) = load_err {
+                if let Some(err) = load_err {
                     // log only after the logger has been initialized
-                    log::info!("cannot load config, continuing with defaults: {load_err:?}");
+                    info!(?err, "cannot load config, continuing with defaults");
                 }
                 &*global_config
             })

src/lsp.rs

@@ -28,15 +28,15 @@
 //! - Progress notifications - contains a `token` property which could be used to identify the
 //!   client but the specification also says it has nothing to do with the request IDs
 
-use anyhow::{bail, ensure, Context, Result};
-use serde::Serialize;
-use serde_json::{Map, Value};
 use std::fmt::{self, Debug};
 use std::io::{self, ErrorKind};
 use std::str;
 use std::sync::Arc;
-use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt};
-use tokio::io::{AsyncWrite, AsyncWriteExt};
+
+use anyhow::{bail, ensure, Context, Result};
+use serde::Serialize;
+use serde_json::{Map, Value};
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 /// Every message begins with a HTTP-style header
 ///

src/main.rs

@@ -29,11 +29,7 @@ struct ClientArgs {
 }
 
 #[derive(Args, Debug)]
-struct ServerArgs {
-    /// Dump all communication with the client
-    #[arg(long)]
-    dump: bool,
-}
+struct ServerArgs {}
 
 #[derive(Subcommand, Debug)]
 enum Cmd {
@@ -49,7 +45,7 @@ async fn main() -> anyhow::Result<()> {
     let cli = Cli::parse();
 
     match cli.command {
-        Some(Cmd::Server(args)) => server::main(args.dump).await,
+        Some(Cmd::Server(_args)) => server::main().await,
         Some(Cmd::Client(args)) => client::main(args.server_path, args.server_args).await,
         None => {
             let server_path = env::var("RA_MUX_SERVER").unwrap_or_else(|_| "rust-analyzer".into());

src/proto.rs

@@ -1,6 +1,7 @@
+use std::env;
+
 use anyhow::{ensure, Context, Result};
 use serde_derive::{Deserialize, Serialize};
-use std::env;
 use tokio::io::{AsyncBufRead, AsyncBufReadExt};
 
 #[derive(Serialize, Deserialize)]

src/server.rs

@@ -6,21 +6,20 @@
 //! LSP Multiplexer attempts to solve this problem by spawning a single rust-analyzer instance per
 //! cargo workspace and routing the messages through TCP to multiple clients.
 
-use crate::server::client::Client;
-use crate::server::instance::InstanceRegistry;
 use anyhow::{Context, Result};
 use ra_multiplex::config::Config;
 use tokio::net::TcpListener;
 use tokio::task;
+use tracing::{debug, error, info, info_span, warn, Instrument};
+
+use crate::server::client::Client;
+use crate::server::instance::InstanceRegistry;
 
 mod async_once_cell;
 mod client;
 mod instance;
 
-pub async fn main(dump: bool) -> Result<()> {
-    if dump {
-        todo!("--dump is not yet implemented");
-    }
+pub async fn main() -> Result<()> {
     let config = Config::load_or_default().await;
     let registry = InstanceRegistry::new().await;
 
@@ -31,18 +30,21 @@ pub async fn main(dump: bool) -> Result<()> {
                 let port = addr.port();
                 let registry = registry.clone();
 
-                log::info!("[{port}] client connected");
-                task::spawn(async move {
-                    match Client::process(socket, port, registry).await {
-                        Ok(_) => log::debug!("[{port}] client initialized"),
-                        Err(err) => log::error!("[{port}] client error: {err:?}"),
+                task::spawn(
+                    async move {
+                        info!("client connected");
+                        match Client::process(socket, port, registry).await {
+                            Ok(_) => debug!("client initialized"),
+                            Err(err) => error!("client error: {err:?}"),
+                        }
                     }
-                });
+                    .instrument(info_span!("client", %port)),
+                );
             }
             Err(err) => match err.kind() {
                 // ignore benign errors
                 std::io::ErrorKind::NotConnected => {
-                    log::warn!("listener error {err}");
+                    warn!("listener error {err}");
                 }
                 _ => {
                     Err(err).context("accept connection")?;