4 files changed
180 )180 )
181 };181 };
182182
183 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));183 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("noq_udp=warn,info"));
184184
185 let tree = {185 let tree = {
186 #[cfg(feature = "indicatif")]186 #[cfg(feature = "indicatif")]
--- a/remowt/cmds/remowt-ssh/src/main.rs
+++ b/remowt/cmds/remowt-ssh/src/main.rs
@@ -18,10 +18,10 @@
use remowt_ui_prompt::{PrependSourcePrompter, Source};
use tokio::io::unix::AsyncFd;
use tokio::io::{AsyncRead, ReadBuf};
-use tokio::signal::unix::{signal, SignalKind};
+use tokio::signal::unix::{SignalKind, signal};
use tracing::debug;
+use tracing_subscriber::EnvFilter;
use tracing_subscriber::prelude::*;
-use tracing_subscriber::EnvFilter;
#[derive(Parser)]
enum Opts {
--- a/remowt/crates/remowt-client/src/lib.rs
+++ b/remowt/crates/remowt-client/src/lib.rs
@@ -27,7 +27,7 @@
fs,
io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader},
};
-use tracing::{debug, info, warn};
+use tracing::{Instrument as _, debug, info, warn};
use uuid::Uuid;
pub mod editor;
@@ -655,28 +655,31 @@
context: String,
stderr: bool,
) -> JoinHandle<()> {
- tokio::spawn(async move {
- let mut reader = BufReader::new(stream);
- let mut buf = Vec::with_capacity(4096);
- loop {
- buf.clear();
- match reader.read_until(b'\n', &mut buf).await {
- Ok(0) => break,
- Ok(_) => {
- let line = String::from_utf8_lossy(buf.strip_suffix(b"\n").unwrap_or(&buf));
- if stderr {
- warn!(context = %context, "{line}");
- } else {
- info!(context = %context, "{line}");
+ tokio::spawn(
+ async move {
+ let mut reader = BufReader::new(stream);
+ let mut buf = Vec::with_capacity(4096);
+ loop {
+ buf.clear();
+ match reader.read_until(b'\n', &mut buf).await {
+ Ok(0) => break,
+ Ok(_) => {
+ let line = String::from_utf8_lossy(buf.strip_suffix(b"\n").unwrap_or(&buf));
+ if stderr {
+ warn!(context = %context, "{line}");
+ } else {
+ info!(context = %context, "{line}");
+ }
+ }
+ Err(e) => {
+ warn!(context = %context, "child stdio read failed: {e}");
+ break;
}
- }
- Err(e) => {
- warn!(context = %context, "child stdio read failed: {e}");
- break;
}
}
}
- })
+ .in_current_span(),
+ )
}
fn local_runtime_dir() -> Result<(Utf8PathBuf, Option<TempDir>)> {
--- a/remowt/crates/remowt-endpoints/src/nix_daemon.rs
+++ b/remowt/crates/remowt-endpoints/src/nix_daemon.rs
@@ -1,11 +1,12 @@
use std::process::Stdio;
use std::sync::Arc;
+use bifrostlink::Config;
use bifrostlink::declarative::endpoints;
-use bifrostlink::Config;
use remowt_link_shared::iroh_tunnel::{TunnelAddr, TunnelDialer};
use serde::{Deserialize, Serialize};
use std::result::Result;
+use tokio::io::{self, AsyncWriteExt as _};
use tokio::process::Command;
#[derive(Clone)]
@@ -47,11 +48,16 @@
let mut stdin = child.stdin.take().expect("piped");
let mut stdout = child.stdout.take().expect("piped");
tokio::spawn(async move {
- let (mut tr, mut tw) = tokio::io::split(tunnel);
- let _ = tokio::join!(
- tokio::io::copy(&mut tr, &mut stdin),
- tokio::io::copy(&mut stdout, &mut tw),
- );
+ let (mut tr, mut tw) = io::split(tunnel);
+ let to_child = async {
+ let _ = io::copy(&mut tr, &mut stdin).await;
+ let _ = stdin.shutdown().await;
+ };
+ let from_child = async {
+ let _ = io::copy(&mut stdout, &mut tw).await;
+ let _ = tw.shutdown().await;
+ };
+ tokio::join!(to_child, from_child);
let _ = child.wait().await;
});
Ok(())