git.iko.soy/helix
on master help sign in
Initial debug adapter protocol implementation
0f6e81bDmitry Sharshakov
Cargo.lock+12
compare with master
unifiedsplit
@@ -322,6 +322,18 @@ dependencies = [
322 "unicode-width",
322 "unicode-width",
323]
323]
324
324
325+[[package]]
326+name = "helix-dap"
327+version = "0.3.0"
328+dependencies = [
329+ "anyhow",
330+ "log",
331+ "serde",
332+ "serde_json",
333+ "thiserror",
334+ "tokio",
335+]
336+
325[[package]]
337[[package]]
326name = "helix-lsp"
338name = "helix-lsp"
327version = "0.4.1"
339version = "0.4.1"
Cargo.toml+1
@@ -6,6 +6,7 @@ members = [
6 "helix-tui",
6 "helix-tui",
7 "helix-syntax",
7 "helix-syntax",
8 "helix-lsp",
8 "helix-lsp",
9+ "helix-dap"
9]
10]
10
11
11# Build helix-syntax in release mode to make the code path faster in development.
12# Build helix-syntax in release mode to make the code path faster in development.
helix-dap/Cargo.tomlnew file+20
@@ -0,0 +1,20 @@
1+[package]
2+name = "helix-dap"
3+version = "0.3.0"
4+authors = ["Blaž Hrastnik <blaz@mxxn.io>"]
5+edition = "2018"
6+license = "MPL-2.0"
7+description = "DAP client implementation for Helix project"
8+categories = ["editor"]
9+repository = "https://github.com/helix-editor/helix"
10+homepage = "https://helix-editor.com"
11+
12+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
13+
14+[dependencies]
15+anyhow = "1.0"
16+log = "0.4"
17+serde = { version = "1.0", features = ["derive"] }
18+serde_json = "1.0"
19+thiserror = "1.0"
20+tokio = { version = "1.9", features = ["rt", "rt-multi-thread", "io-util", "io-std", "time", "process", "macros", "fs", "parking_lot"] }
helix-dap/examples/dap-basic.rsnew file+51
@@ -0,0 +1,51 @@
1+use helix_dap::{Client, Result, SourceBreakpoint};
2+
3+#[tokio::main]
4+pub async fn main() -> Result<()> {
5+ let mut client = Client::start("nc", vec!["127.0.0.1", "7777"], 0)?;
6+
7+ println!("init: {:?}", client.initialize().await);
8+ println!("caps: {:?}", client.capabilities());
9+ println!(
10+ "launch: {:?}",
11+ client.launch("/tmp/godebug/main".to_owned()).await
12+ );
13+
14+ println!(
15+ "breakpoints: {:?}",
16+ client
17+ .set_breakpoints(
18+ "/tmp/godebug/main.go".to_owned(),
19+ vec![SourceBreakpoint {
20+ line: 6,
21+ column: Some(2),
22+ }]
23+ )
24+ .await
25+ );
26+
27+ let mut _in = String::new();
28+ std::io::stdin()
29+ .read_line(&mut _in)
30+ .expect("Failed to read line");
31+
32+ println!("configurationDone: {:?}", client.configuration_done().await);
33+ println!("stopped: {:?}", client.wait_for_stopped().await);
34+ println!("stack trace: {:?}", client.stack_trace(1).await);
35+
36+ let mut _in = String::new();
37+ std::io::stdin()
38+ .read_line(&mut _in)
39+ .expect("Failed to read line");
40+
41+ println!("continued: {:?}", client.continue_thread(0).await);
42+
43+ let mut _in = String::new();
44+ std::io::stdin()
45+ .read_line(&mut _in)
46+ .expect("Failed to read line");
47+
48+ println!("disconnect: {:?}", client.disconnect().await);
49+
50+ Ok(())
51+}
helix-dap/src/client.rsnew file+352
@@ -0,0 +1,352 @@
1+use crate::{
2+ transport::{Event, Payload, Request, Response, Transport},
3+ Result,
4+};
5+use serde::{Deserialize, Serialize};
6+use serde_json::{from_value, to_value, Value};
7+use std::process::Stdio;
8+use std::sync::atomic::{AtomicU64, Ordering};
9+use tokio::{
10+ io::{BufReader, BufWriter},
11+ process::{Child, Command},
12+ sync::mpsc::{channel, UnboundedReceiver, UnboundedSender},
13+};
14+
15+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
16+#[serde(rename_all = "camelCase")]
17+pub struct DebuggerCapabilities {
18+ supports_configuration_done_request: bool,
19+ supports_function_breakpoints: bool,
20+ supports_conditional_breakpoints: bool,
21+ supports_exception_info_request: bool,
22+ support_terminate_debuggee: bool,
23+ supports_delayed_stack_trace_loading: bool,
24+}
25+
26+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
27+#[serde(rename_all = "camelCase")]
28+struct InitializeArguments {
29+ client_id: String,
30+ client_name: String,
31+ adapter_id: String,
32+ locale: String,
33+ #[serde(rename = "linesStartAt1")]
34+ lines_start_at_one: bool,
35+ #[serde(rename = "columnsStartAt1")]
36+ columns_start_at_one: bool,
37+ path_format: String,
38+ supports_variable_type: bool,
39+ supports_variable_paging: bool,
40+ supports_run_in_terminal_request: bool,
41+ supports_memory_references: bool,
42+ supports_progress_reporting: bool,
43+ supports_invalidated_event: bool,
44+}
45+
46+// TODO: split out
47+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
48+#[serde(rename_all = "camelCase")]
49+struct LaunchArguments {
50+ mode: String,
51+ program: String,
52+}
53+
54+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
55+#[serde(rename_all = "camelCase")]
56+pub struct Source {
57+ path: Option<String>,
58+}
59+
60+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
61+#[serde(rename_all = "camelCase")]
62+pub struct SourceBreakpoint {
63+ pub line: usize,
64+ pub column: Option<usize>,
65+}
66+
67+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
68+#[serde(rename_all = "camelCase")]
69+struct SetBreakpointsArguments {
70+ source: Source,
71+ breakpoints: Option<Vec<SourceBreakpoint>>,
72+}
73+
74+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
75+#[serde(rename_all = "camelCase")]
76+pub struct Breakpoint {
77+ pub id: Option<usize>,
78+ pub verified: bool,
79+ pub message: Option<String>,
80+ pub source: Option<Source>,
81+ pub line: Option<usize>,
82+ pub column: Option<usize>,
83+ pub end_line: Option<usize>,
84+ pub end_column: Option<usize>,
85+ pub instruction_reference: Option<String>,
86+ pub offset: Option<usize>,
87+}
88+
89+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
90+#[serde(rename_all = "camelCase")]
91+struct SetBreakpointsResponseBody {
92+ breakpoints: Option<Vec<Breakpoint>>,
93+}
94+
95+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
96+#[serde(rename_all = "camelCase")]
97+struct ContinueArguments {
98+ thread_id: usize,
99+}
100+
101+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
102+#[serde(rename_all = "camelCase")]
103+struct ContinueResponseBody {
104+ all_threads_continued: Option<bool>,
105+}
106+
107+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
108+#[serde(rename_all = "camelCase")]
109+struct StackFrameFormat {
110+ parameters: Option<bool>,
111+ parameter_types: Option<bool>,
112+ parameter_names: Option<bool>,
113+ parameter_values: Option<bool>,
114+ line: Option<bool>,
115+ module: Option<bool>,
116+ include_all: Option<bool>,
117+}
118+
119+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
120+#[serde(rename_all = "camelCase")]
121+struct StackTraceArguments {
122+ thread_id: usize,
123+ start_frame: Option<usize>,
124+ levels: Option<usize>,
125+ format: Option<StackFrameFormat>,
126+}
127+
128+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
129+#[serde(rename_all = "camelCase")]
130+pub struct StackFrame {
131+ id: usize,
132+ name: String,
133+ source: Option<Source>,
134+ line: usize,
135+ column: usize,
136+ end_line: Option<usize>,
137+ end_column: Option<usize>,
138+ can_restart: Option<bool>,
139+ instruction_pointer_reference: Option<String>,
140+ // module_id
141+ presentation_hint: Option<String>,
142+}
143+
144+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
145+#[serde(rename_all = "camelCase")]
146+struct StackTraceResponseBody {
147+ total_frames: Option<usize>,
148+ stack_frames: Vec<StackFrame>,
149+}
150+
151+#[derive(Debug)]
152+pub struct Client {
153+ id: usize,
154+ _process: Child,
155+ server_tx: UnboundedSender<Request>,
156+ server_rx: UnboundedReceiver<Payload>,
157+ request_counter: AtomicU64,
158+ capabilities: Option<DebuggerCapabilities>,
159+}
160+
161+impl Client {
162+ pub fn start(cmd: &str, args: Vec<&str>, id: usize) -> Result<Self> {
163+ let process = Command::new(cmd)
164+ .args(args)
165+ .stdin(Stdio::piped())
166+ .stdout(Stdio::piped())
167+ // make sure the process is reaped on drop
168+ .kill_on_drop(true)
169+ .spawn();
170+
171+ let mut process = process?;
172+
173+ // TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?
174+ let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));
175+ let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));
176+
177+ let (server_rx, server_tx) = Transport::start(reader, writer, id);
178+
179+ let client = Self {
180+ id,
181+ _process: process,
182+ server_tx,
183+ server_rx,
184+ request_counter: AtomicU64::new(0),
185+ capabilities: None,
186+ };
187+
188+ // TODO: async client.initialize()
189+ // maybe use an arc<atomic> flag
190+
191+ Ok(client)
192+ }
193+
194+ pub fn id(&self) -> usize {
195+ self.id
196+ }
197+
198+ fn next_request_id(&self) -> u64 {
199+ self.request_counter.fetch_add(1, Ordering::Relaxed)
200+ }
201+
202+ async fn request(&self, command: String, arguments: Option<Value>) -> Result<Response> {
203+ let (callback_rx, mut callback_tx) = channel(1);
204+
205+ let req = Request {
206+ back_ch: Some(callback_rx),
207+ seq: self.next_request_id(),
208+ msg_type: "request".to_owned(),
209+ command,
210+ arguments,
211+ };
212+
213+ self.server_tx
214+ .send(req)
215+ .expect("Failed to send request to debugger");
216+
217+ callback_tx
218+ .recv()
219+ .await
220+ .expect("Failed to receive response")
221+ }
222+
223+ pub fn capabilities(&self) -> &DebuggerCapabilities {
224+ self.capabilities
225+ .as_ref()
226+ .expect("language server not yet initialized!")
227+ }
228+
229+ pub async fn initialize(&mut self) -> Result<()> {
230+ let args = InitializeArguments {
231+ client_id: "hx".to_owned(),
232+ client_name: "helix".to_owned(),
233+ adapter_id: "go".to_owned(),
234+ locale: "en-us".to_owned(),
235+ lines_start_at_one: true,
236+ columns_start_at_one: true,
237+ path_format: "path".to_owned(),
238+ supports_variable_type: false,
239+ supports_variable_paging: false,
240+ supports_run_in_terminal_request: false,
241+ supports_memory_references: false,
242+ supports_progress_reporting: true,
243+ supports_invalidated_event: true,
244+ };
245+
246+ let response = self
247+ .request("initialize".to_owned(), to_value(args).ok())
248+ .await?;
249+ self.capabilities = from_value(response.body.unwrap()).ok();
250+
251+ Ok(())
252+ }
253+
254+ pub async fn disconnect(&mut self) -> Result<()> {
255+ self.request("disconnect".to_owned(), None).await?;
256+ Ok(())
257+ }
258+
259+ pub async fn launch(&mut self, executable: String) -> Result<()> {
260+ let args = LaunchArguments {
261+ mode: "exec".to_owned(),
262+ program: executable,
263+ };
264+
265+ self.request("launch".to_owned(), to_value(args).ok())
266+ .await?;
267+
268+ match self
269+ .server_rx
270+ .recv()
271+ .await
272+ .expect("Expected initialized event")
273+ {
274+ Payload::Event(Event { event, .. }) => {
275+ if event == "initialized".to_owned() {
276+ Ok(())
277+ } else {
278+ unreachable!()
279+ }
280+ }
281+ _ => unreachable!(),
282+ }
283+ }
284+
285+ pub async fn set_breakpoints(
286+ &mut self,
287+ file: String,
288+ breakpoints: Vec<SourceBreakpoint>,
289+ ) -> Result<Option<Vec<Breakpoint>>> {
290+ let args = SetBreakpointsArguments {
291+ source: Source { path: Some(file) },
292+ breakpoints: Some(breakpoints),
293+ };
294+
295+ let response = self
296+ .request("setBreakpoints".to_owned(), to_value(args).ok())
297+ .await?;
298+ let body: Option<SetBreakpointsResponseBody> = from_value(response.body.unwrap()).ok();
299+
300+ Ok(body.map(|b| b.breakpoints).unwrap())
301+ }
302+
303+ pub async fn configuration_done(&mut self) -> Result<()> {
304+ self.request("configurationDone".to_owned(), None).await?;
305+ Ok(())
306+ }
307+
308+ pub async fn wait_for_stopped(&mut self) -> Result<()> {
309+ match self.server_rx.recv().await.expect("Expected stopped event") {
310+ Payload::Event(Event { event, .. }) => {
311+ if event == "stopped".to_owned() {
312+ Ok(())
313+ } else {
314+ unreachable!()
315+ }
316+ }
317+ _ => unreachable!(),
318+ }
319+ }
320+
321+ pub async fn continue_thread(&mut self, thread_id: usize) -> Result<Option<bool>> {
322+ let args = ContinueArguments { thread_id };
323+
324+ let response = self
325+ .request("continue".to_owned(), to_value(args).ok())
326+ .await?;
327+
328+ let body: Option<ContinueResponseBody> = from_value(response.body.unwrap()).ok();
329+
330+ Ok(body.map(|b| b.all_threads_continued).unwrap())
331+ }
332+
333+ pub async fn stack_trace(
334+ &mut self,
335+ thread_id: usize,
336+ ) -> Result<(Vec<StackFrame>, Option<usize>)> {
337+ let args = StackTraceArguments {
338+ thread_id,
339+ start_frame: None,
340+ levels: None,
341+ format: None,
342+ };
343+
344+ let response = self
345+ .request("stackTrace".to_owned(), to_value(args).ok())
346+ .await?;
347+
348+ let body: StackTraceResponseBody = from_value(response.body.unwrap()).unwrap();
349+
350+ Ok((body.stack_frames, body.total_frames))
351+ }
352+}
helix-dap/src/lib.rsnew file+21
@@ -0,0 +1,21 @@
1+mod client;
2+mod transport;
3+
4+pub use client::{Breakpoint, Client, SourceBreakpoint};
5+pub use transport::{Event, Payload, Request, Response, Transport};
6+
7+use thiserror::Error;
8+#[derive(Error, Debug)]
9+pub enum Error {
10+ #[error("failed to parse: {0}")]
11+ Parse(#[from] serde_json::Error),
12+ #[error("IO Error: {0}")]
13+ IO(#[from] std::io::Error),
14+ #[error("request timed out")]
15+ Timeout,
16+ #[error("server closed the stream")]
17+ StreamClosed,
18+ #[error(transparent)]
19+ Other(#[from] anyhow::Error),
20+}
21+pub type Result<T> = core::result::Result<T, Error>;
helix-dap/src/transport.rsnew file+282
@@ -0,0 +1,282 @@
1+use crate::{Error, Result};
2+use anyhow::Context;
3+use log::error;
4+use serde::{Deserialize, Serialize};
5+use serde_json::Value;
6+use std::collections::HashMap;
7+use std::sync::Arc;
8+use tokio::{
9+ io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
10+ process::{ChildStdin, ChildStdout},
11+ sync::{
12+ mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
13+ Mutex,
14+ },
15+};
16+
17+#[derive(Debug, Clone, Deserialize, Serialize)]
18+pub struct Request {
19+ #[serde(skip)]
20+ pub back_ch: Option<Sender<Result<Response>>>,
21+ pub seq: u64,
22+ #[serde(rename = "type")]
23+ pub msg_type: String,
24+ pub command: String,
25+ pub arguments: Option<Value>,
26+}
27+
28+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
29+pub struct Response {
30+ pub seq: u64,
31+ #[serde(rename = "type")]
32+ pub msg_type: String,
33+ pub request_seq: u64,
34+ pub success: bool,
35+ pub command: String,
36+ pub message: Option<String>,
37+ pub body: Option<Value>,
38+}
39+
40+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
41+pub struct Event {
42+ pub seq: u64,
43+ #[serde(rename = "type")]
44+ pub msg_type: String,
45+ pub event: String,
46+ pub body: Option<Value>,
47+}
48+
49+#[derive(Debug, Clone, Deserialize, Serialize)]
50+#[serde(untagged)]
51+pub enum Payload {
52+ // type = "event"
53+ Event(Event),
54+ // type = "response"
55+ Response(Response),
56+ // type = "request"
57+ Request(Request),
58+}
59+
60+#[derive(Debug)]
61+pub struct Transport {
62+ id: usize,
63+ pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>,
64+}
65+
66+impl Transport {
67+ pub fn start(
68+ server_stdout: BufReader<ChildStdout>,
69+ server_stdin: BufWriter<ChildStdin>,
70+ id: usize,
71+ ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) {
72+ let (client_tx, rx) = unbounded_channel();
73+ let (tx, client_rx) = unbounded_channel();
74+
75+ let transport = Self {
76+ id,
77+ pending_requests: Mutex::new(HashMap::default()),
78+ };
79+
80+ let transport = Arc::new(transport);
81+
82+ tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));
83+ tokio::spawn(Self::send(transport, server_stdin, client_rx));
84+
85+ (rx, tx)
86+ }
87+
88+ async fn recv_server_message(
89+ reader: &mut (impl AsyncBufRead + Unpin + Send),
90+ buffer: &mut String,
91+ ) -> Result<Payload> {
92+ let mut content_length = None;
93+ loop {
94+ buffer.truncate(0);
95+ reader.read_line(buffer).await?;
96+ let header = buffer.trim();
97+
98+ if header.is_empty() {
99+ break;
100+ }
101+
102+ let mut parts = header.split(": ");
103+
104+ match (parts.next(), parts.next(), parts.next()) {
105+ (Some("Content-Length"), Some(value), None) => {
106+ content_length = Some(value.parse().context("invalid content length")?);
107+ }
108+ (Some(_), Some(_), None) => {}
109+ _ => {
110+ return Err(std::io::Error::new(
111+ std::io::ErrorKind::Other,
112+ "Failed to parse header",
113+ )
114+ .into());
115+ }
116+ }
117+ }
118+
119+ let content_length = content_length.context("missing content length")?;
120+
121+ //TODO: reuse vector
122+ let mut content = vec![0; content_length];
123+ reader.read_exact(&mut content).await?;
124+ let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;
125+
126+ // TODO: `info!` here
127+ println!("<- DAP {}", msg);
128+
129+ // try parsing as output (server response) or call (server request)
130+ let output: serde_json::Result<Payload> = serde_json::from_str(msg);
131+
132+ Ok(output?)
133+ }
134+
135+ async fn send_payload_to_server(
136+ &self,
137+ server_stdin: &mut BufWriter<ChildStdin>,
138+ req: Request,
139+ ) -> Result<()> {
140+ let json = serde_json::to_string(&req)?;
141+ match req.back_ch {
142+ Some(back) => {
143+ self.pending_requests.lock().await.insert(req.seq, back);
144+ ()
145+ }
146+ None => {}
147+ }
148+ self.send_string_to_server(server_stdin, json).await
149+ }
150+
151+ async fn send_string_to_server(
152+ &self,
153+ server_stdin: &mut BufWriter<ChildStdin>,
154+ request: String,
155+ ) -> Result<()> {
156+ // TODO: `info!` here
157+ println!("-> DAP {}", request);
158+
159+ // send the headers
160+ server_stdin
161+ .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
162+ .await?;
163+
164+ // send the body
165+ server_stdin.write_all(request.as_bytes()).await?;
166+
167+ server_stdin.flush().await?;
168+
169+ Ok(())
170+ }
171+
172+ async fn process_server_message(
173+ &self,
174+ client_tx: &UnboundedSender<Payload>,
175+ msg: Payload,
176+ ) -> Result<()> {
177+ let (id, result) = match msg {
178+ Payload::Response(Response {
179+ success: true,
180+ seq,
181+ request_seq,
182+ ..
183+ }) => {
184+ // TODO: `info!` here
185+ println!("<- DAP success ({}, in response to {})", seq, request_seq);
186+ if let Payload::Response(val) = msg {
187+ (request_seq, Ok(val))
188+ } else {
189+ unreachable!();
190+ }
191+ }
192+ Payload::Response(Response {
193+ success: false,
194+ message,
195+ body,
196+ request_seq,
197+ command,
198+ ..
199+ }) => {
200+ // TODO: `error!` here
201+ println!(
202+ "<- DAP error {:?} ({:?}) for command #{} {}",
203+ message, body, request_seq, command
204+ );
205+ (
206+ request_seq,
207+ Err(Error::Other(anyhow::format_err!("{:?}", body))),
208+ )
209+ }
210+ Payload::Request(Request {
211+ ref command,
212+ ref seq,
213+ ..
214+ }) => {
215+ // TODO: `info!` here
216+ println!("<- DAP request {} #{}", command, seq);
217+ client_tx.send(msg).expect("Failed to send");
218+ return Ok(());
219+ }
220+ Payload::Event(Event {
221+ ref event, ref seq, ..
222+ }) => {
223+ // TODO: `info!` here
224+ println!("<- DAP event {} #{}", event, seq);
225+ client_tx.send(msg).expect("Failed to send");
226+ return Ok(());
227+ }
228+ };
229+
230+ let tx = self
231+ .pending_requests
232+ .lock()
233+ .await
234+ .remove(&id)
235+ .expect("pending_request with id not found!");
236+
237+ match tx.send(result).await {
238+ Ok(_) => (),
239+ Err(_) => error!(
240+ "Tried sending response into a closed channel (id={:?}), original request likely timed out",
241+ id
242+ ),
243+ };
244+
245+ Ok(())
246+ }
247+
248+ async fn recv(
249+ transport: Arc<Self>,
250+ mut server_stdout: BufReader<ChildStdout>,
251+ client_tx: UnboundedSender<Payload>,
252+ ) {
253+ let mut recv_buffer = String::new();
254+ loop {
255+ match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
256+ Ok(msg) => {
257+ transport
258+ .process_server_message(&client_tx, msg)
259+ .await
260+ .unwrap();
261+ }
262+ Err(err) => {
263+ error!("err: <- {:?}", err);
264+ break;
265+ }
266+ }
267+ }
268+ }
269+
270+ async fn send(
271+ transport: Arc<Self>,
272+ mut server_stdin: BufWriter<ChildStdin>,
273+ mut client_rx: UnboundedReceiver<Request>,
274+ ) {
275+ while let Some(req) = client_rx.recv().await {
276+ transport
277+ .send_payload_to_server(&mut server_stdin, req)
278+ .await
279+ .unwrap()
280+ }
281+ }
282+}