@@ -5,11 +5,18 @@ use crate::{
55use log::{error, info};
66use serde::{Deserialize, Serialize};
77use serde_json::{from_value, to_value, Value};
8−use std::sync::{
9− atomic::{AtomicU64, Ordering},
10− Arc,
8+use std::{
9+ collections::HashMap,
10+ net::{IpAddr, Ipv4Addr},
11+ process::Stdio,
12+};
13+use std::{
14+ net::SocketAddr,
15+ sync::{
16+ atomic::{AtomicU64, Ordering},
17+ Arc,
18+ },
1119};
12−use std::{collections::HashMap, process::Stdio};
1320use tokio::{
1421 io::{AsyncBufRead, AsyncWrite, BufReader, BufWriter},
1522 join,
@@ -19,6 +26,7 @@ use tokio::{
1926 mpsc::{channel, Receiver, Sender, UnboundedReceiver, UnboundedSender},
2027 Mutex,
2128 },
29+ time,
2230};
2331
2432#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
@@ -376,6 +384,82 @@ impl Client {
376384 Ok(client)
377385 }
378386
387+ pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> {
388+ let stream = TcpStream::connect(addr).await?;
389+ let (rx, tx) = stream.into_split();
390+ Self::streams(Box::new(BufReader::new(rx)), Box::new(tx), id, None)
391+ }
392+
393+ pub fn stdio(cmd: &str, args: Vec<&str>, id: usize) -> Result<Self> {
394+ let process = Command::new(cmd)
395+ .args(args)
396+ .stdin(Stdio::piped())
397+ .stdout(Stdio::piped())
398+ // make sure the process is reaped on drop
399+ .kill_on_drop(true)
400+ .spawn();
401+
402+ let mut process = process?;
403+
404+ // TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?
405+ let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));
406+ let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));
407+
408+ Self::streams(
409+ Box::new(BufReader::new(reader)),
410+ Box::new(writer),
411+ id,
412+ Some(process),
413+ )
414+ }
415+
416+ async fn get_port() -> Option<u16> {
417+ Some(
418+ tokio::net::TcpListener::bind(SocketAddr::new(
419+ IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
420+ 0,
421+ ))
422+ .await
423+ .ok()?
424+ .local_addr()
425+ .ok()?
426+ .port(),
427+ )
428+ }
429+
430+ pub async fn tcp_process(
431+ cmd: &str,
432+ args: Vec<&str>,
433+ port_format: &str,
434+ id: usize,
435+ ) -> Result<Self> {
436+ let port = Self::get_port().await.unwrap();
437+
438+ let process = Command::new(cmd)
439+ .args(args)
440+ .args(port_format.replace("{}", &port.to_string()).split(' '))
441+ // make sure the process is reaped on drop
442+ .kill_on_drop(true)
443+ .spawn()?;
444+
445+ // Wait for adapter to become ready for connection
446+ time::sleep(time::Duration::from_millis(500)).await;
447+
448+ let stream = TcpStream::connect(SocketAddr::new(
449+ IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
450+ port,
451+ ))
452+ .await?;
453+
454+ let (rx, tx) = stream.into_split();
455+ Self::streams(
456+ Box::new(BufReader::new(rx)),
457+ Box::new(tx),
458+ id,
459+ Some(process),
460+ )
461+ }
462+
379463 async fn recv(
380464 awaited_events: Arc<Mutex<HashMap<String, Sender<Event>>>>,
381465 mut server_rx: UnboundedReceiver<Payload>,
@@ -413,35 +497,6 @@ impl Client {
413497 tx
414498 }
415499
416− pub async fn tcp(addr: std::net::SocketAddr, id: usize) -> Result<Self> {
417− let stream = TcpStream::connect(addr).await?;
418− let (rx, tx) = stream.into_split();
419− Self::streams(Box::new(BufReader::new(rx)), Box::new(tx), id, None)
420− }
421−
422− pub fn stdio(cmd: &str, args: Vec<&str>, id: usize) -> Result<Self> {
423− let process = Command::new(cmd)
424− .args(args)
425− .stdin(Stdio::piped())
426− .stdout(Stdio::piped())
427− // make sure the process is reaped on drop
428− .kill_on_drop(true)
429− .spawn();
430−
431− let mut process = process?;
432−
433− // TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?
434− let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));
435− let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));
436−
437− Self::streams(
438− Box::new(BufReader::new(reader)),
439− Box::new(writer),
440− id,
441− Some(process),
442− )
443− }
444−
445500 pub fn id(&self) -> usize {
446501 self.id
447502 }