@@ -1,15 +1,9 @@
1−use std::collections::HashMap;
2−use std::io;
3−
4−use log::{error, info};
5−
6−use crate::Error;
7−
8−type Result<T> = core::result::Result<T, Error>;
9−
1+use crate::Result;
102use jsonrpc_core as jsonrpc;
3+use log::{error, info};
4+use serde::{Deserialize, Serialize};
115use serde_json::Value;
12−
6+use std::collections::HashMap;
137use tokio::{
148 io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
159 process::{ChildStderr, ChildStdin, ChildStdout},
@@ -26,12 +20,11 @@ pub enum Payload {
2620 Response(jsonrpc::Output),
2721}
2822
29−use serde::{Deserialize, Serialize};
3023/// A type representing all possible values sent from the server to the client.
3124#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
3225#[serde(deny_unknown_fields)]
3326#[serde(untagged)]
34−enum Message {
27+enum ServerMessage {
3528 /// A regular JSON-RPC request output (single response).
3629 Output(jsonrpc::Output),
3730 /// A JSON-RPC request or notification.
@@ -39,34 +32,32 @@ enum Message {
3932}
4033
4134pub struct Transport {
42− incoming: UnboundedSender<jsonrpc::Call>,
43− outgoing: UnboundedReceiver<Payload>,
35+ client_tx: UnboundedSender<jsonrpc::Call>,
36+ client_rx: UnboundedReceiver<Payload>,
4437
4538 pending_requests: HashMap<jsonrpc::Id, Sender<Result<Value>>>,
46− headers: HashMap<String, String>,
4739
48− writer: BufWriter<ChildStdin>,
49− reader: BufReader<ChildStdout>,
50− stderr: BufReader<ChildStderr>,
40+ server_stdin: BufWriter<ChildStdin>,
41+ server_stdout: BufReader<ChildStdout>,
42+ server_stderr: BufReader<ChildStderr>,
5143}
5244
5345impl Transport {
5446 pub fn start(
55− reader: BufReader<ChildStdout>,
56− writer: BufWriter<ChildStdin>,
57− stderr: BufReader<ChildStderr>,
47+ server_stdout: BufReader<ChildStdout>,
48+ server_stdin: BufWriter<ChildStdin>,
49+ server_stderr: BufReader<ChildStderr>,
5850 ) -> (UnboundedReceiver<jsonrpc::Call>, UnboundedSender<Payload>) {
59− let (incoming, rx) = unbounded_channel();
60− let (tx, outgoing) = unbounded_channel();
51+ let (client_tx, rx) = unbounded_channel();
52+ let (tx, client_rx) = unbounded_channel();
6153
6254 let transport = Self {
63− reader,
64− writer,
65− stderr,
66− incoming,
67− outgoing,
55+ server_stdout,
56+ server_stdin,
57+ server_stderr,
58+ client_tx,
59+ client_rx,
6860 pending_requests: HashMap::default(),
69− headers: HashMap::default(),
7061 };
7162
7263 tokio::spawn(transport.duplex());
@@ -74,105 +65,104 @@ impl Transport {
7465 (rx, tx)
7566 }
7667
77− async fn recv(
68+ async fn recv_server_message(
7869 reader: &mut (impl AsyncBufRead + Unpin + Send),
79− headers: &mut HashMap<String, String>,
80− ) -> core::result::Result<Message, std::io::Error> {
81− // read headers
70+ buffer: &mut String,
71+ ) -> Result<ServerMessage> {
72+ let mut content_length = None;
8273 loop {
83− let mut header = String::new();
84− // detect pipe closed if 0
85− reader.read_line(&mut header).await?;
86− let header = header.trim();
74+ buffer.truncate(0);
75+ reader.read_line(buffer).await?;
76+ let header = buffer.trim();
8777
8878 if header.is_empty() {
8979 break;
9080 }
9181
92− let parts: Vec<&str> = header.split(": ").collect();
93− if parts.len() != 2 {
94− return Err(std::io::Error::new(
95− std::io::ErrorKind::Other,
96− "Failed to parse header",
97− ));
82+ let mut parts = header.split(": ");
83+
84+ match (parts.next(), parts.next(), parts.next()) {
85+ (Some("Content-Length"), Some(value), None) => {
86+ content_length = Some(value.parse().unwrap());
87+ }
88+ (Some(_), Some(_), None) => {}
89+ _ => {
90+ return Err(std::io::Error::new(
91+ std::io::ErrorKind::Other,
92+ "Failed to parse header",
93+ )
94+ .into());
95+ }
9896 }
99− headers.insert(parts[0].to_string(), parts[1].to_string());
10097 }
10198
102− // find content-length
103− let content_length = headers.get("Content-Length").unwrap().parse().unwrap();
99+ let content_length = content_length.unwrap();
104100
101+ //TODO: reuse vector
105102 let mut content = vec![0; content_length];
106103 reader.read_exact(&mut content).await?;
107104 let msg = String::from_utf8(content).unwrap();
108105
109− // read data
110−
111106 info!("<- {}", msg);
112107
113108 // try parsing as output (server response) or call (server request)
114− let output: serde_json::Result<Message> = serde_json::from_str(&msg);
109+ let output: serde_json::Result<ServerMessage> = serde_json::from_str(&msg);
115110
116111 Ok(output?)
117112 }
118113
119− async fn err(
114+ async fn recv_server_error(
120115 err: &mut (impl AsyncBufRead + Unpin + Send),
121− ) -> core::result::Result<(), std::io::Error> {
122− let mut line = String::new();
123− err.read_line(&mut line).await?;
124− error!("err <- {}", line);
116+ buffer: &mut String,
117+ ) -> Result<()> {
118+ buffer.truncate(0);
119+ err.read_line(buffer).await?;
120+ error!("err <- {}", buffer);
125121
126122 Ok(())
127123 }
128124
129− pub async fn send_payload(&mut self, payload: Payload) -> io::Result<()> {
130− match payload {
125+ async fn send_payload_to_server(&mut self, payload: Payload) -> Result<()> {
126+ //TODO: reuse string
127+ let json = match payload {
131128 Payload::Request { chan, value } => {
132129 self.pending_requests.insert(value.id.clone(), chan);
133−
134− let json = serde_json::to_string(&value)?;
135− self.send(json).await
136− }
137− Payload::Notification(value) => {
138− let json = serde_json::to_string(&value)?;
139− self.send(json).await
130+ serde_json::to_string(&value)?
140131 }
141− Payload::Response(error) => {
142− let json = serde_json::to_string(&error)?;
143− self.send(json).await
144− }
145− }
132+ Payload::Notification(value) => serde_json::to_string(&value)?,
133+ Payload::Response(error) => serde_json::to_string(&error)?,
134+ };
135+ self.send_string_to_server(json).await
146136 }
147137
148− pub async fn send(&mut self, request: String) -> io::Result<()> {
138+ async fn send_string_to_server(&mut self, request: String) -> Result<()> {
149139 info!("-> {}", request);
150140
151141 // send the headers
152− self.writer
142+ self.server_stdin
153143 .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())
154144 .await?;
155145
156146 // send the body
157− self.writer.write_all(request.as_bytes()).await?;
147+ self.server_stdin.write_all(request.as_bytes()).await?;
158148
159− self.writer.flush().await?;
149+ self.server_stdin.flush().await?;
160150
161151 Ok(())
162152 }
163153
164− async fn recv_msg(&mut self, msg: Message) -> anyhow::Result<()> {
154+ async fn process_server_message(&mut self, msg: ServerMessage) -> Result<()> {
165155 match msg {
166− Message::Output(output) => self.recv_response(output).await?,
167− Message::Call(call) => {
168− self.incoming.send(call).unwrap();
156+ ServerMessage::Output(output) => self.process_request_response(output).await?,
157+ ServerMessage::Call(call) => {
158+ self.client_tx.send(call).unwrap();
169159 // let notification = Notification::parse(&method, params);
170160 }
171161 };
172162 Ok(())
173163 }
174164
175− async fn recv_response(&mut self, output: jsonrpc::Output) -> io::Result<()> {
165+ async fn process_request_response(&mut self, output: jsonrpc::Output) -> Result<()> {
176166 let (id, result) = match output {
177167 jsonrpc::Output::Success(jsonrpc::Success { id, result, .. }) => {
178168 info!("<- {}", result);
@@ -200,29 +190,33 @@ impl Transport {
200190 Ok(())
201191 }
202192
203− pub async fn duplex(mut self) {
193+ async fn duplex(mut self) {
194+ let mut recv_buffer = String::new();
195+ let mut err_buffer = String::new();
204196 loop {
205197 tokio::select! {
206198 // client -> server
207− msg = self.outgoing.recv() => {
208− if msg.is_none() {
209− break;
199+ msg = self.client_rx.recv() => {
200+ match msg {
201+ Some(msg) => {
202+ self.send_payload_to_server(msg).await.unwrap()
203+ },
204+ None => break
210205 }
211− let msg = msg.unwrap();
212−
213− self.send_payload(msg).await.unwrap();
214206 }
215− // server <- client
216− msg = Self::recv(&mut self.reader, &mut self.headers) => {
217− if msg.is_err() {
218− error!("err: <- {:?}", msg);
219− break;
207+ // server -> client
208+ msg = Self::recv_server_message(&mut self.server_stdout, &mut recv_buffer) => {
209+ match msg {
210+ Ok(msg) => {
211+ self.process_server_message(msg).await.unwrap();
212+ }
213+ Err(_) => {
214+ error!("err: <- {:?}", msg);
215+ break;
216+ },
220217 }
221− let msg = msg.unwrap();
222−
223− self.recv_msg(msg).await.unwrap();
224218 }
225− _msg = Self::err(&mut self.stderr) => {}
219+ _msg = Self::recv_server_error(&mut self.server_stderr, &mut err_buffer) => {}
226220 }
227221 }
228222 }