26a55dc · helix · git.iko.soy

1−use crate::{Error, Result};

1+use crate::{Error, Result};

2−use anyhow::Context;

2+use anyhow::Context;

3−use log::{error, info};

3+use log::{error, info};

4−use serde::{Deserialize, Serialize};

4+use serde::{Deserialize, Serialize};

5−use serde_json::Value;

5+use serde_json::Value;

6−use std::collections::HashMap;

6+use std::collections::HashMap;

7−use std::sync::Arc;

7+use std::sync::Arc;

8−use tokio::{

8+use tokio::{

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

9+ io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt},

10− process::{ChildStdin, ChildStdout},

11− sync::{

10+ sync::{

12− mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},

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

13− Mutex,

12+ Mutex,

14− },

13+ },

15−};

14+};

16−

15+

17−#[derive(Debug, Clone, Deserialize, Serialize)]

16+#[derive(Debug, Clone, Deserialize, Serialize)]

18−pub struct Request {

17+pub struct Request {

19− #[serde(skip)]

18+ #[serde(skip)]

20− pub back_ch: Option<Sender<Result<Response>>>,

19+ pub back_ch: Option<Sender<Result<Response>>>,

21− pub seq: u64,

20+ pub seq: u64,

22− #[serde(rename = "type")]

21+ #[serde(rename = "type")]

23− pub msg_type: String,

22+ pub msg_type: String,

24− pub command: String,

23+ pub command: String,

25− pub arguments: Option<Value>,

24+ pub arguments: Option<Value>,

26−}

25+}

27−

26+

28−#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]

27+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]

29−pub struct Response {

28+pub struct Response {

30− pub seq: u64,

29+ pub seq: u64,

31− #[serde(rename = "type")]

30+ #[serde(rename = "type")]

32− pub msg_type: String,

31+ pub msg_type: String,

33− pub request_seq: u64,

32+ pub request_seq: u64,

34− pub success: bool,

33+ pub success: bool,

35− pub command: String,

34+ pub command: String,

36− pub message: Option<String>,

35+ pub message: Option<String>,

37− pub body: Option<Value>,

36+ pub body: Option<Value>,

38−}

37+}

39−

38+

40−#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]

39+#[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]

41−pub struct Event {

40+pub struct Event {

42− pub seq: u64,

41+ pub seq: u64,

43− #[serde(rename = "type")]

42+ #[serde(rename = "type")]

44− pub msg_type: String,

43+ pub msg_type: String,

45− pub event: String,

44+ pub event: String,

46− pub body: Option<Value>,

45+ pub body: Option<Value>,

47−}

46+}

48−

47+

49−#[derive(Debug, Clone, Deserialize, Serialize)]

48+#[derive(Debug, Clone, Deserialize, Serialize)]

50−#[serde(untagged)]

49+#[serde(untagged)]

51−pub enum Payload {

50+pub enum Payload {

52− // type = "event"

51+ // type = "event"

53− Event(Event),

52+ Event(Event),

54− // type = "response"

53+ // type = "response"

55− Response(Response),

54+ Response(Response),

56− // type = "request"

55+ // type = "request"

57− Request(Request),

56+ Request(Request),

58−}

57+}

59−

58+

60−#[derive(Debug)]

59+#[derive(Debug)]

61−pub struct Transport {

60+pub struct Transport {

62− id: usize,

61+ id: usize,

63− pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>,

62+ pending_requests: Mutex<HashMap<u64, Sender<Result<Response>>>>,

64−}

63+}

65−

64+

66−impl Transport {

65+impl Transport {

67− pub fn start(

66+ pub fn start(

68− server_stdout: BufReader<ChildStdout>,

67+ server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,

69− server_stdin: BufWriter<ChildStdin>,

68+ server_stdin: Box<dyn AsyncWrite + Unpin + Send>,

70− id: usize,

69+ id: usize,

71− ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) {

70+ ) -> (UnboundedReceiver<Payload>, UnboundedSender<Request>) {

72− let (client_tx, rx) = unbounded_channel();

71+ let (client_tx, rx) = unbounded_channel();

73− let (tx, client_rx) = unbounded_channel();

72+ let (tx, client_rx) = unbounded_channel();

74−

73+

75− let transport = Self {

74+ let transport = Self {

76− id,

75+ id,

77− pending_requests: Mutex::new(HashMap::default()),

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

78− };

77+ };

79−

78+

80− let transport = Arc::new(transport);

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

81−

80+

82− tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));

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

83− tokio::spawn(Self::send(transport, server_stdin, client_rx));

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

84−

83+

85− (rx, tx)

84+ (rx, tx)

86− }

85+ }

87−

86+

88− async fn recv_server_message(

87+ async fn recv_server_message(

89− reader: &mut (impl AsyncBufRead + Unpin + Send),

88+ reader: &mut Box<dyn AsyncBufRead + Unpin + Send>,

90− buffer: &mut String,

89+ buffer: &mut String,

91− ) -> Result<Payload> {

90+ ) -> Result<Payload> {

92− let mut content_length = None;

91+ let mut content_length = None;

93− loop {

92+ loop {

94− buffer.truncate(0);

93+ buffer.truncate(0);

95− reader.read_line(buffer).await?;

94+ reader.read_line(buffer).await?;

96− let header = buffer.trim();

95+ let header = buffer.trim();

97−

96+

98− if header.is_empty() {

97+ if header.is_empty() {

99− break;

98+ break;

100− }

99+ }

101−

100+

102− let mut parts = header.split(": ");

101+ let mut parts = header.split(": ");

103−

102+

104− match (parts.next(), parts.next(), parts.next()) {

103+ match (parts.next(), parts.next(), parts.next()) {

105− (Some("Content-Length"), Some(value), None) => {

104+ (Some("Content-Length"), Some(value), None) => {

106− content_length = Some(value.parse().context("invalid content length")?);

105+ content_length = Some(value.parse().context("invalid content length")?);

107− }

106+ }

108− (Some(_), Some(_), None) => {}

107+ (Some(_), Some(_), None) => {}

109− _ => {

108+ _ => {

110− return Err(std::io::Error::new(

109+ return Err(std::io::Error::new(

111− std::io::ErrorKind::Other,

110+ std::io::ErrorKind::Other,

112− "Failed to parse header",

111+ "Failed to parse header",

113− )

112+ )

114− .into());

113+ .into());

115− }

114+ }

116− }

115+ }

117− }

116+ }

118−

117+

119− let content_length = content_length.context("missing content length")?;

118+ let content_length = content_length.context("missing content length")?;

120−

119+

121− //TODO: reuse vector

120+ //TODO: reuse vector

122− let mut content = vec![0; content_length];

121+ let mut content = vec![0; content_length];

123− reader.read_exact(&mut content).await?;

122+ reader.read_exact(&mut content).await?;

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

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

125−

124+

126− info!("<- DAP {}", msg);

125+ info!("<- DAP {}", msg);

127−

126+

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

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

129− let output: serde_json::Result<Payload> = serde_json::from_str(msg);

128+ let output: serde_json::Result<Payload> = serde_json::from_str(msg);

130−

129+

131− Ok(output?)

130+ Ok(output?)

132− }

131+ }

133−

132+

134− async fn send_payload_to_server(

133+ async fn send_payload_to_server(

135− &self,

134+ &self,

136− server_stdin: &mut BufWriter<ChildStdin>,

135+ server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,

137− req: Request,

136+ req: Request,

138− ) -> Result<()> {

137+ ) -> Result<()> {

139− let json = serde_json::to_string(&req)?;

138+ let json = serde_json::to_string(&req)?;

140− if let Some(back) = req.back_ch {

139+ if let Some(back) = req.back_ch {

141− self.pending_requests.lock().await.insert(req.seq, back);

140+ self.pending_requests.lock().await.insert(req.seq, back);

142− }

141+ }

143− self.send_string_to_server(server_stdin, json).await

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

144− }

143+ }

145−

144+

146− async fn send_string_to_server(

145+ async fn send_string_to_server(

147− &self,

146+ &self,

148− server_stdin: &mut BufWriter<ChildStdin>,

147+ server_stdin: &mut Box<dyn AsyncWrite + Unpin + Send>,

149− request: String,

148+ request: String,

150− ) -> Result<()> {

149+ ) -> Result<()> {

151− info!("-> DAP {}", request);

150+ info!("-> DAP {}", request);

152−

151+

153− // send the headers

152+ // send the headers

154− server_stdin

153+ server_stdin

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

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

156− .await?;

155+ .await?;

157−

156+

158− // send the body

157+ // send the body

159− server_stdin.write_all(request.as_bytes()).await?;

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

160−

159+

161− server_stdin.flush().await?;

160+ server_stdin.flush().await?;

162−

161+

163− Ok(())

162+ Ok(())

164− }

163+ }

165−

164+

166− async fn process_server_message(

165+ async fn process_server_message(

167− &self,

166+ &self,

168− client_tx: &UnboundedSender<Payload>,

167+ client_tx: &UnboundedSender<Payload>,

169− msg: Payload,

168+ msg: Payload,

170− ) -> Result<()> {

169+ ) -> Result<()> {

171− let (id, result) = match msg {

170+ let (id, result) = match msg {

172− Payload::Response(Response {

171+ Payload::Response(Response {

173− success: true,

172+ success: true,

174− seq,

173+ seq,

175− request_seq,

174+ request_seq,

176− ..

175+ ..

177− }) => {

176+ }) => {

178− info!("<- DAP success ({}, in response to {})", seq, request_seq);

177+ info!("<- DAP success ({}, in response to {})", seq, request_seq);

179− if let Payload::Response(val) = msg {

178+ if let Payload::Response(val) = msg {

180− (request_seq, Ok(val))

179+ (request_seq, Ok(val))

181− } else {

180+ } else {

182− unreachable!();

181+ unreachable!();

183− }

182+ }

184− }

183+ }

185− Payload::Response(Response {

184+ Payload::Response(Response {

186− success: false,

185+ success: false,

187− message,

186+ message,

188− body,

187+ body,

189− request_seq,

188+ request_seq,

190− command,

189+ command,

191− ..

190+ ..

192− }) => {

191+ }) => {

193− error!(

192+ error!(

194− "<- DAP error {:?} ({:?}) for command #{} {}",

193+ "<- DAP error {:?} ({:?}) for command #{} {}",

195− message, body, request_seq, command

194+ message, body, request_seq, command

196− );

195+ );

197− (

196+ (

198− request_seq,

197+ request_seq,

199− Err(Error::Other(anyhow::format_err!("{:?}", body))),

198+ Err(Error::Other(anyhow::format_err!("{:?}", body))),

200− )

199+ )

201− }

200+ }

202− Payload::Request(Request {

201+ Payload::Request(Request {

203− ref command,

202+ ref command,

204− ref seq,

203+ ref seq,

205− ..

204+ ..

206− }) => {

205+ }) => {

207− info!("<- DAP request {} #{}", command, seq);

206+ info!("<- DAP request {} #{}", command, seq);

208− client_tx.send(msg).expect("Failed to send");

207+ client_tx.send(msg).expect("Failed to send");

209− return Ok(());

208+ return Ok(());

210− }

209+ }

211− Payload::Event(Event {

210+ Payload::Event(Event {

212− ref event, ref seq, ..

211+ ref event, ref seq, ..

213− }) => {

212+ }) => {

214− info!("<- DAP event {} #{}", event, seq);

213+ info!("<- DAP event {} #{}", event, seq);

215− client_tx.send(msg).expect("Failed to send");

214+ client_tx.send(msg).expect("Failed to send");

216− return Ok(());

215+ return Ok(());

217− }

216+ }

218− };

217+ };

219−

218+

220− let tx = self

219+ let tx = self

221− .pending_requests

220+ .pending_requests

222− .lock()

221+ .lock()

223− .await

222+ .await

224− .remove(&id)

223+ .remove(&id)

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

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

226−

225+

227− match tx.send(result).await {

226+ match tx.send(result).await {

228− Ok(_) => (),

227+ Ok(_) => (),

229− Err(_) => error!(

228+ Err(_) => error!(

230− "Tried sending response into a closed channel (id={:?}), original request likely timed out",

229+ "Tried sending response into a closed channel (id={:?}), original request likely timed out",

231− id

230+ id

232− ),

231+ ),

233− };

232+ };

234−

233+

235− Ok(())

234+ Ok(())

236− }

235+ }

237−

236+

238− async fn recv(

237+ async fn recv(

239− transport: Arc<Self>,

238+ transport: Arc<Self>,

240− mut server_stdout: BufReader<ChildStdout>,

239+ mut server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,

241− client_tx: UnboundedSender<Payload>,

240+ client_tx: UnboundedSender<Payload>,

242− ) {

241+ ) {

243− let mut recv_buffer = String::new();

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

244− loop {

243+ loop {

245− match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {

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

246− Ok(msg) => {

245+ Ok(msg) => {

247− transport

246+ transport

248− .process_server_message(&client_tx, msg)

247+ .process_server_message(&client_tx, msg)

249− .await

248+ .await

250− .unwrap();

249+ .unwrap();

251− }

250+ }

252− Err(err) => {

251+ Err(err) => {

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

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

254− break;

253+ break;

255− }

254+ }

256− }

255+ }

257− }

256+ }

258− }

257+ }

259−

258+

260− async fn send(

259+ async fn send(

261− transport: Arc<Self>,

260+ transport: Arc<Self>,

262− mut server_stdin: BufWriter<ChildStdin>,

261+ mut server_stdin: Box<dyn AsyncWrite + Unpin + Send>,

263− mut client_rx: UnboundedReceiver<Request>,

262+ mut client_rx: UnboundedReceiver<Request>,

264− ) {

263+ ) {

265− while let Some(req) = client_rx.recv().await {

264+ while let Some(req) = client_rx.recv().await {

266− transport

265+ transport

267− .send_payload_to_server(&mut server_stdin, req)

266+ .send_payload_to_server(&mut server_stdin, req)

268− .await

267+ .await

269− .unwrap()

268+ .unwrap()

270− }

269+ }

271− }

270+ }

272−}

271+}