385a6b5 · helix · git.iko.soy

5use serde::{Deserialize, Serialize};

6use serde_json::Value;

7use std::collections::HashMap;

8+use std::sync::Arc;

8use tokio::{

9use tokio::{

9 io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},

10 io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},

10 process::{ChildStderr, ChildStdin, ChildStdout},

11 process::{ChildStderr, ChildStdin, ChildStdout},

12+ sync::{

11− sync::mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},

13+ mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},

14+ Mutex,

15+ },

12};

16};

13

17

14#[derive(Debug)]

18#[derive(Debug)]

35#[derive(Debug)]

39#[derive(Debug)]

36pub struct Transport {

40pub struct Transport {

37 id: usize,

41 id: usize,

38− client_tx: UnboundedSender<(usize, jsonrpc::Call)>,

39− client_rx: UnboundedReceiver<Payload>,

40−

41− pending_requests: HashMap<jsonrpc::Id, Sender<Result<Value>>>,

42+ pending_requests: Mutex<HashMap<jsonrpc::Id, Sender<Result<Value>>>>,

42−

43− server_stdin: BufWriter<ChildStdin>,

44− server_stdout: BufReader<ChildStdout>,

45− server_stderr: BufReader<ChildStderr>,

46}

43}

47

44

48impl Transport {

45impl Transport {

60

57

61 let transport = Self {

58 let transport = Self {

62 id,

59 id,

63− server_stdout,

64− server_stdin,

65− server_stderr,

66− client_tx,

67− client_rx,

68− pending_requests: HashMap::default(),

60+ pending_requests: Mutex::new(HashMap::default()),

69 };

61 };

70

62

63+ let transport = Arc::new(transport);

64+

65+ tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));

71− tokio::spawn(transport.duplex());

66+ tokio::spawn(Self::err(transport.clone(), server_stderr));

67+ tokio::spawn(Self::send(transport, server_stdin, client_rx));

72

68

73 (rx, tx)

69 (rx, tx)

74 }

70 }

109 //TODO: reuse vector

105 //TODO: reuse vector

110 let mut content = vec![0; content_length];

106 let mut content = vec![0; content_length];

111 reader.read_exact(&mut content).await?;

107 reader.read_exact(&mut content).await?;

112− let msg = String::from_utf8(content).context("invalid utf8 from server")?;

108+ let msg = std::str::from_utf8(&content).context("invalid utf8 from server")?;

113

109

114 info!("<- {}", msg);

110 info!("<- {}", msg);

115

111

116 // try parsing as output (server response) or call (server request)

112 // try parsing as output (server response) or call (server request)

117− let output: serde_json::Result<ServerMessage> = serde_json::from_str(&msg);

113+ let output: serde_json::Result<ServerMessage> = serde_json::from_str(msg);

118

114

119 Ok(output?)

115 Ok(output?)

120 }

116 }

130 Ok(())

126 Ok(())

131 }

127 }

132

128

133− async fn send_payload_to_server(&mut self, payload: Payload) -> Result<()> {

129+ async fn send_payload_to_server(

130+ &self,

131+ server_stdin: &mut BufWriter<ChildStdin>,

132+ payload: Payload,

133+ ) -> Result<()> {

134 //TODO: reuse string

135 let json = match payload {

136 Payload::Request { chan, value } => {

137+ self.pending_requests

138+ .lock()

139+ .await

137− self.pending_requests.insert(value.id.clone(), chan);

140+ .insert(value.id.clone(), chan);

138 serde_json::to_string(&value)?

141 serde_json::to_string(&value)?

139 }

142 }

140 Payload::Notification(value) => serde_json::to_string(&value)?,

143 Payload::Notification(value) => serde_json::to_string(&value)?,

141 Payload::Response(error) => serde_json::to_string(&error)?,

144 Payload::Response(error) => serde_json::to_string(&error)?,

142 };

145 };

143− self.send_string_to_server(json).await

146+ self.send_string_to_server(server_stdin, json).await

144 }

147 }

145

148

146− async fn send_string_to_server(&mut self, request: String) -> Result<()> {

149+ async fn send_string_to_server(

150+ &self,

151+ server_stdin: &mut BufWriter<ChildStdin>,

152+ request: String,

153+ ) -> Result<()> {

147 info!("-> {}", request);

154 info!("-> {}", request);

148

155

149 // send the headers

156 // send the headers

150− self.server_stdin

157+ server_stdin

151 .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())

158 .write_all(format!("Content-Length: {}\r\n\r\n", request.len()).as_bytes())

152 .await?;

159 .await?;

153

160

154 // send the body

161 // send the body

155− self.server_stdin.write_all(request.as_bytes()).await?;

162+ server_stdin.write_all(request.as_bytes()).await?;

156

163

157− self.server_stdin.flush().await?;

164+ server_stdin.flush().await?;

158

165

159 Ok(())

166 Ok(())

160 }

167 }

161

168

162− async fn process_server_message(&mut self, msg: ServerMessage) -> Result<()> {

169+ async fn process_server_message(

170+ &self,

171+ client_tx: &UnboundedSender<(usize, jsonrpc::Call)>,

172+ msg: ServerMessage,

173+ ) -> Result<()> {

163 match msg {

174 match msg {

164 ServerMessage::Output(output) => self.process_request_response(output).await?,

175 ServerMessage::Output(output) => self.process_request_response(output).await?,

165 ServerMessage::Call(call) => {

176 ServerMessage::Call(call) => {

166− self.client_tx

177+ client_tx

167 .send((self.id, call))

178 .send((self.id, call))

168 .context("failed to send a message to server")?;

179 .context("failed to send a message to server")?;

169 // let notification = Notification::parse(&method, params);

180 // let notification = Notification::parse(&method, params);

172 Ok(())

183 Ok(())

173 }

184 }

174

185

175− async fn process_request_response(&mut self, output: jsonrpc::Output) -> Result<()> {

186+ async fn process_request_response(&self, output: jsonrpc::Output) -> Result<()> {

176 let (id, result) = match output {

187 let (id, result) = match output {

177 jsonrpc::Output::Success(jsonrpc::Success { id, result, .. }) => {

188 jsonrpc::Output::Success(jsonrpc::Success { id, result, .. }) => {

178 info!("<- {}", result);

189 info!("<- {}", result);

186

197

187 let tx = self

198 let tx = self

188 .pending_requests

199 .pending_requests

200+ .lock()

201+ .await

189 .remove(&id)

202 .remove(&id)

190 .expect("pending_request with id not found!");

203 .expect("pending_request with id not found!");

191

204

200 Ok(())

213 Ok(())

201 }

214 }

202

215

203− async fn duplex(mut self) {

216+ async fn recv(

217+ transport: Arc<Self>,

218+ mut server_stdout: BufReader<ChildStdout>,

219+ client_tx: UnboundedSender<(usize, jsonrpc::Call)>,

220+ ) {

204 let mut recv_buffer = String::new();

221 let mut recv_buffer = String::new();

205− let mut err_buffer = String::new();

206 loop {

222 loop {

207− tokio::select! {

223+ match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {

208− // client -> server

209− msg = self.client_rx.recv() => {

210− match msg {

211− Some(msg) => {

224+ Ok(msg) => {

212− self.send_payload_to_server(msg).await.unwrap()

225+ transport

213− },

226+ .process_server_message(&client_tx, msg)

214− None => break

227+ .await

215− }

228+ .unwrap();

216 }

229 }

217− // server -> client

218− msg = Self::recv_server_message(&mut self.server_stdout, &mut recv_buffer) => {

219− match msg {

220− Ok(msg) => {

221− self.process_server_message(msg).await.unwrap();

222− }

223− Err(_) => {

230+ Err(err) => {

224− error!("err: <- {:?}", msg);

231+ error!("err: <- {:?}", err);

225− break;

232+ break;

226− },

227− }

228 }

233 }

229− _msg = Self::recv_server_error(&mut self.server_stderr, &mut err_buffer) => {}

230 }

234 }

231 }

235 }

232 }

236 }

237+

238+ async fn err(_transport: Arc<Self>, mut server_stderr: BufReader<ChildStderr>) {

239+ let mut recv_buffer = String::new();

240+ loop {

241+ match Self::recv_server_error(&mut server_stderr, &mut recv_buffer).await {

242+ Ok(_) => {}

243+ Err(err) => {

244+ error!("err: <- {:?}", err);

245+ break;

246+ }

247+ }

248+ }

249+ }

250+

251+ async fn send(

252+ transport: Arc<Self>,

253+ mut server_stdin: BufWriter<ChildStdin>,

254+ mut client_rx: UnboundedReceiver<Payload>,

255+ ) {

256+ while let Some(msg) = client_rx.recv().await {

257+ transport

258+ .send_payload_to_server(&mut server_stdin, msg)

259+ .await

260+ .unwrap()

261+ }

262+ }

233}

263}