ea66670 · helix · git.iko.soy

git.iko.soy/helix

on master help sign in

helix-lsp cleanup

ea66670Egor Karavaev5 years ago

Cargo.lock−16

compare with master

unifiedsplit

@@ -237,12 +237,6 @@ dependencies = [

237 "wasi",

238]

239

240−[[package]]

241−name = "glob"

242−version = "0.3.0"

243−source = "registry+https://github.com/rust-lang/crates.io-index"

244−checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574"

245−

246[[package]]

240[[package]]

247name = "globset"

241name = "globset"

248version = "0.4.6"

242version = "0.4.6"

@@ -282,19 +276,15 @@ dependencies = [

282 "anyhow",

276 "anyhow",

283 "futures-executor",

277 "futures-executor",

284 "futures-util",

278 "futures-util",

285− "glob",

286 "helix-core",

279 "helix-core",

287 "jsonrpc-core",

280 "jsonrpc-core",

288 "log",

281 "log",

289 "lsp-types",

282 "lsp-types",

290− "once_cell",

291− "pathdiff",

292 "serde",

283 "serde",

293 "serde_json",

284 "serde_json",

294 "thiserror",

285 "thiserror",

295 "tokio",

286 "tokio",

296 "tokio-stream",

287 "tokio-stream",

297− "url",

298]

288]

299

289

300[[package]]

290[[package]]

@@ -596,12 +586,6 @@ dependencies = [

596 "winapi",

586 "winapi",

597]

587]

598

588

599−[[package]]

600−name = "pathdiff"

601−version = "0.2.0"

602−source = "registry+https://github.com/rust-lang/crates.io-index"

603−checksum = "877630b3de15c0b64cc52f659345724fbf6bdad9bd9566699fc53688f3c34a34"

604−

605[[package]]

589[[package]]

606name = "percent-encoding"

590name = "percent-encoding"

607version = "2.1.0"

591version = "2.1.0"

helix-lsp/Cargo.toml+8−13

@@ -10,19 +10,14 @@ license = "MPL-2.0"

10[dependencies]

11helix-core = { path = "../helix-core" }

13−once_cell = "1.4"

13+anyhow = "1.0"

14−

14+futures-executor = "0.3"

15+futures-util = { version = "0.3", features = ["std", "async-await"], default-features = false }

16+jsonrpc-core = { version = "17.1", default-features = false } # don't pull in all of futures

17+log = "0.4"

15lsp-types = { version = "0.89", features = ["proposed"] }

18lsp-types = { version = "0.89", features = ["proposed"] }

16−tokio = { version = "1", features = ["full"] }

17−tokio-stream = "0.1.5"

18−futures-executor = { version = "0.3" }

19−url = "2"

20−pathdiff = "0.2"

21−glob = "0.3"

22−anyhow = "1"

23−serde_json = "1.0"

24serde = { version = "1.0", features = ["derive"] }

19serde = { version = "1.0", features = ["derive"] }

25−jsonrpc-core = { version = "17.1", default-features = false } # don't pull in all of futures

20+serde_json = "1.0"

26−futures-util = { version = "0.3", features = ["std", "async-await"], default-features = false }

27thiserror = "1.0"

21thiserror = "1.0"

28−log = "~0.4"

22+tokio = { version = "1.6", features = ["full"] }

23+tokio-stream = "0.1.6"

helix-lsp/src/client.rs+20−41

@@ -4,30 +4,22 @@ use crate::{

4};

6use helix_core::{find_root, ChangeSet, Rope};

7−

8−// use std::collections::HashMap;

9−use std::future::Future;

10−use std::sync::atomic::{AtomicU64, Ordering};

11−

12use jsonrpc_core as jsonrpc;

7use jsonrpc_core as jsonrpc;

13use lsp_types as lsp;

8use lsp_types as lsp;

14use serde_json::Value;

9use serde_json::Value;

15−

10+use std::future::Future;

16use std::process::Stdio;

11use std::process::Stdio;

12+use std::sync::atomic::{AtomicU64, Ordering};

17use tokio::{

13use tokio::{

18 io::{BufReader, BufWriter},

14 io::{BufReader, BufWriter},

19− // prelude::*,

20 process::{Child, Command},

15 process::{Child, Command},

21 sync::mpsc::{channel, UnboundedReceiver, UnboundedSender},

16 sync::mpsc::{channel, UnboundedReceiver, UnboundedSender},

22};

17};

24pub struct Client {

19pub struct Client {

25 _process: Child,

20 _process: Child,

26−

27− outgoing: UnboundedSender<Payload>,

21+ server_tx: UnboundedSender<Payload>,

28− // pub incoming: Receiver<Call>,

29− pub request_counter: AtomicU64,

22+ request_counter: AtomicU64,

30−

31 capabilities: Option<lsp::ServerCapabilities>,

23 capabilities: Option<lsp::ServerCapabilities>,

32 offset_encoding: OffsetEncoding,

24 offset_encoding: OffsetEncoding,

33}

25}

@@ -43,40 +35,27 @@ impl Client {

43 .kill_on_drop(true)

35 .kill_on_drop(true)

44 .spawn();

36 .spawn();

46− // use std::io::ErrorKind;

47− let mut process = match process {

38+ let mut process = process?;

48− Ok(process) => process,

49− Err(err) => match err.kind() {

50− // ErrorKind::NotFound | ErrorKind::PermissionDenied => {

51− // return Err(Error::Other(err.into()))

52− // }

53− _kind => return Err(Error::Other(err.into())),

54− },

55− };

57 // TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?

40 // TODO: do we need bufreader/writer here? or do we use async wrappers on unblock?

58 let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));

41 let writer = BufWriter::new(process.stdin.take().expect("Failed to open stdin"));

59 let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));

42 let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));

60 let stderr = BufReader::new(process.stderr.take().expect("Failed to open stderr"));

43 let stderr = BufReader::new(process.stderr.take().expect("Failed to open stderr"));

62− let (incoming, outgoing) = Transport::start(reader, writer, stderr);

45+ let (server_rx, server_tx) = Transport::start(reader, writer, stderr);

64 let client = Self {

47 let client = Self {

65 _process: process,

48 _process: process,

66−

49+ server_tx,

67− outgoing,

68− // incoming,

69 request_counter: AtomicU64::new(0),

50 request_counter: AtomicU64::new(0),

70−

71 capabilities: None,

51 capabilities: None,

72− // diagnostics: HashMap::new(),

73 offset_encoding: OffsetEncoding::Utf8,

52 offset_encoding: OffsetEncoding::Utf8,

74 };

53 };

76 // TODO: async client.initialize()

55 // TODO: async client.initialize()

77 // maybe use an arc<atomic> flag

56 // maybe use an arc<atomic> flag

79− Ok((client, incoming))

58+ Ok((client, server_rx))

80 }

59 }

82 fn next_request_id(&self) -> jsonrpc::Id {

61 fn next_request_id(&self) -> jsonrpc::Id {

@@ -106,7 +85,7 @@ impl Client {

106 }

85 }

107

108 /// Execute a RPC request on the language server.

87 /// Execute a RPC request on the language server.

109− pub async fn request<R: lsp::request::Request>(&self, params: R::Params) -> Result<R::Result>

88+ async fn request<R: lsp::request::Request>(&self, params: R::Params) -> Result<R::Result>

110 where

89 where

111 R::Params: serde::Serialize,

90 R::Params: serde::Serialize,

112 R::Result: core::fmt::Debug, // TODO: temporary

91 R::Result: core::fmt::Debug, // TODO: temporary

@@ -118,17 +97,20 @@ impl Client {

118 }

97 }

119

120 /// Execute a RPC request on the language server.

99 /// Execute a RPC request on the language server.

121− pub fn call<R: lsp::request::Request>(

100+ fn call<R: lsp::request::Request>(

122 &self,

101 &self,

123 params: R::Params,

102 params: R::Params,

124 ) -> impl Future<Output = Result<Value>>

103 ) -> impl Future<Output = Result<Value>>

125 where

104 where

126 R::Params: serde::Serialize,

105 R::Params: serde::Serialize,

127 {

106 {

128− let outgoing = self.outgoing.clone();

107+ let server_tx = self.server_tx.clone();

129 let id = self.next_request_id();

108 let id = self.next_request_id();

130

109

131 async move {

110 async move {

111+ use std::time::Duration;

112+ use tokio::time::timeout;

113+

132 let params = serde_json::to_value(params)?;

114 let params = serde_json::to_value(params)?;

133

115

134 let request = jsonrpc::MethodCall {

116 let request = jsonrpc::MethodCall {

@@ -140,16 +122,13 @@ impl Client {

140

122

141 let (tx, mut rx) = channel::<Result<Value>>(1);

123 let (tx, mut rx) = channel::<Result<Value>>(1);

142

124

143− outgoing

125+ server_tx

144 .send(Payload::Request {

126 .send(Payload::Request {

145 chan: tx,

127 chan: tx,

146 value: request,

128 value: request,

147 })

129 })

148 .map_err(|e| Error::Other(e.into()))?;

130 .map_err(|e| Error::Other(e.into()))?;

149

131

150− use std::time::Duration;

151− use tokio::time::timeout;

152−

153 timeout(Duration::from_secs(2), rx.recv())

132 timeout(Duration::from_secs(2), rx.recv())

154 .await

133 .await

155 .map_err(|_| Error::Timeout)? // return Timeout

134 .map_err(|_| Error::Timeout)? // return Timeout

@@ -158,14 +137,14 @@ impl Client {

158 }

137 }

159

138

160 /// Send a RPC notification to the language server.

139 /// Send a RPC notification to the language server.

161− pub fn notify<R: lsp::notification::Notification>(

140+ fn notify<R: lsp::notification::Notification>(

162 &self,

141 &self,

163 params: R::Params,

142 params: R::Params,

164 ) -> impl Future<Output = Result<()>>

143 ) -> impl Future<Output = Result<()>>

165 where

144 where

166 R::Params: serde::Serialize,

145 R::Params: serde::Serialize,

167 {

146 {

168− let outgoing = self.outgoing.clone();

147+ let server_tx = self.server_tx.clone();

169

148

170 async move {

149 async move {

171 let params = serde_json::to_value(params)?;

150 let params = serde_json::to_value(params)?;

@@ -176,7 +155,7 @@ impl Client {

176 params: Self::value_into_params(params),

155 params: Self::value_into_params(params),

177 };

156 };

178

157

179− outgoing

158+ server_tx

180 .send(Payload::Notification(notification))

159 .send(Payload::Notification(notification))

181 .map_err(|e| Error::Other(e.into()))?;

160 .map_err(|e| Error::Other(e.into()))?;

182

161

@@ -205,7 +184,7 @@ impl Client {

205 }),

184 }),

206 };

185 };

207

186

208− self.outgoing

187+ self.server_tx

209 .send(Payload::Response(output))

188 .send(Payload::Response(output))

210 .map_err(|e| Error::Other(e.into()))?;

189 .map_err(|e| Error::Other(e.into()))?;

211

190

@@ -216,7 +195,7 @@ impl Client {

216 // General messages

195 // General messages

217 // -------------------------------------------------------------------------------------------

196 // -------------------------------------------------------------------------------------------

218

197

219− pub async fn initialize(&mut self) -> Result<()> {

198+ pub(crate) async fn initialize(&mut self) -> Result<()> {

220 // TODO: delay any requests that are triggered prior to initialize

199 // TODO: delay any requests that are triggered prior to initialize

221 let root = find_root(None).and_then(|root| lsp::Url::from_file_path(root).ok());

200 let root = find_root(None).and_then(|root| lsp::Url::from_file_path(root).ok());

222

201

helix-lsp/src/lib.rs+10−17

@@ -1,28 +1,27 @@

1mod client;

2mod transport;

4−pub use jsonrpc_core as jsonrpc;

5−pub use lsp_types as lsp;

6−

7pub use client::Client;

4pub use client::Client;

5+pub use futures_executor::block_on;

6+pub use jsonrpc::Call;

7+pub use jsonrpc_core as jsonrpc;

8pub use lsp::{Position, Url};

9+pub use lsp_types as lsp;

10−pub type Result<T> = core::result::Result<T, Error>;

11+use futures_util::stream::select_all::SelectAll;

11−

12use helix_core::syntax::LanguageConfiguration;

14−use thiserror::Error;

15−

16use std::{

14use std::{

17 collections::{hash_map::Entry, HashMap},

15 collections::{hash_map::Entry, HashMap},

18 sync::Arc,

16 sync::Arc,

19};

17};

21use serde::{Deserialize, Serialize};

19use serde::{Deserialize, Serialize};

22−

20+use thiserror::Error;

23use tokio_stream::wrappers::UnboundedReceiverStream;

21use tokio_stream::wrappers::UnboundedReceiverStream;

25−pub use futures_executor::block_on;

23+pub type Result<T> = core::result::Result<T, Error>;

24+type LanguageId = String;

27#[derive(Error, Debug)]

26#[derive(Error, Debug)]

28pub enum Error {

27pub enum Error {

@@ -30,6 +29,8 @@ pub enum Error {

30 Rpc(#[from] jsonrpc::Error),

29 Rpc(#[from] jsonrpc::Error),

31 #[error("failed to parse: {0}")]

30 #[error("failed to parse: {0}")]

32 Parse(#[from] serde_json::Error),

31 Parse(#[from] serde_json::Error),

32+ #[error("IO Error: {0}")]

33+ IO(#[from] std::io::Error),

33 #[error("request timed out")]

34 #[error("request timed out")]

34 Timeout,

35 Timeout,

35 #[error("server closed the stream")]

36 #[error("server closed the stream")]

@@ -126,8 +127,6 @@ pub mod util {

126 }),

127 }),

127 )

128 )

128 }

129 }

129−

130− // apply_insert_replace_edit

131}

130}

132

131

133#[derive(Debug, PartialEq, Clone)]

132#[derive(Debug, PartialEq, Clone)]

@@ -173,12 +172,6 @@ impl Notification {

173 }

172 }

174}

173}

175

174

176−pub use jsonrpc::Call;

177−

178−type LanguageId = String;

179−

180−use futures_util::stream::select_all::SelectAll;

181−

182pub struct Registry {

175pub struct Registry {

183 inner: HashMap<LanguageId, Arc<Client>>,

176 inner: HashMap<LanguageId, Arc<Client>>,

184

177

helix-lsp/src/transport.rs+87−93

@@ -1,15 +1,9 @@

1−use std::collections::HashMap;

1+use crate::Result;

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−

10use jsonrpc_core as jsonrpc;

2use jsonrpc_core as jsonrpc;

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

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

11use serde_json::Value;

5use serde_json::Value;

12−

6+use std::collections::HashMap;

13use tokio::{

7use tokio::{

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

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

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

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

@@ -26,12 +20,11 @@ pub enum Payload {

26 Response(jsonrpc::Output),

20 Response(jsonrpc::Output),

27}

21}

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

30/// A type representing all possible values sent from the server to the client.

23/// A type representing all possible values sent from the server to the client.

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

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

32#[serde(deny_unknown_fields)]

25#[serde(deny_unknown_fields)]

33#[serde(untagged)]

26#[serde(untagged)]

34−enum Message {

27+enum ServerMessage {

35 /// A regular JSON-RPC request output (single response).

28 /// A regular JSON-RPC request output (single response).

36 Output(jsonrpc::Output),

29 Output(jsonrpc::Output),

37 /// A JSON-RPC request or notification.

30 /// A JSON-RPC request or notification.

@@ -39,34 +32,32 @@ enum Message {

39}

32}

41pub struct Transport {

34pub struct Transport {

42− incoming: UnboundedSender<jsonrpc::Call>,

35+ client_tx: UnboundedSender<jsonrpc::Call>,

43− outgoing: UnboundedReceiver<Payload>,

36+ client_rx: UnboundedReceiver<Payload>,

45 pending_requests: HashMap<jsonrpc::Id, Sender<Result<Value>>>,

38 pending_requests: HashMap<jsonrpc::Id, Sender<Result<Value>>>,

46− headers: HashMap<String, String>,

48− writer: BufWriter<ChildStdin>,

40+ server_stdin: BufWriter<ChildStdin>,

49− reader: BufReader<ChildStdout>,

41+ server_stdout: BufReader<ChildStdout>,

50− stderr: BufReader<ChildStderr>,

42+ server_stderr: BufReader<ChildStderr>,

51}

43}

53impl Transport {

45impl Transport {

54 pub fn start(

46 pub fn start(

55− reader: BufReader<ChildStdout>,

47+ server_stdout: BufReader<ChildStdout>,

56− writer: BufWriter<ChildStdin>,

48+ server_stdin: BufWriter<ChildStdin>,

57− stderr: BufReader<ChildStderr>,

49+ server_stderr: BufReader<ChildStderr>,

58 ) -> (UnboundedReceiver<jsonrpc::Call>, UnboundedSender<Payload>) {

50 ) -> (UnboundedReceiver<jsonrpc::Call>, UnboundedSender<Payload>) {

59− let (incoming, rx) = unbounded_channel();

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

60− let (tx, outgoing) = unbounded_channel();

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

62 let transport = Self {

54 let transport = Self {

63− reader,

55+ server_stdout,

64− writer,

56+ server_stdin,

65− stderr,

57+ server_stderr,

66− incoming,

58+ client_tx,

67− outgoing,

59+ client_rx,

68 pending_requests: HashMap::default(),

60 pending_requests: HashMap::default(),

69− headers: HashMap::default(),

70 };

61 };

72 tokio::spawn(transport.duplex());

63 tokio::spawn(transport.duplex());

@@ -74,105 +65,104 @@ impl Transport {

74 (rx, tx)

65 (rx, tx)

75 }

66 }

77− async fn recv(

68+ async fn recv_server_message(

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

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

79− headers: &mut HashMap<String, String>,

70+ buffer: &mut String,

80− ) -> core::result::Result<Message, std::io::Error> {

71+ ) -> Result<ServerMessage> {

81− // read headers

72+ let mut content_length = None;

82 loop {

73 loop {

83− let mut header = String::new();

74+ buffer.truncate(0);

84− // detect pipe closed if 0

85− reader.read_line(&mut header).await?;

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

86− let header = header.trim();

76+ let header = buffer.trim();

88 if header.is_empty() {

78 if header.is_empty() {

89 break;

79 break;

90 }

80 }

92− let parts: Vec<&str> = header.split(": ").collect();

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

93− if parts.len() != 2 {

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+ _ => {

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

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

95− std::io::ErrorKind::Other,

91+ std::io::ErrorKind::Other,

96− "Failed to parse header",

92+ "Failed to parse header",

97− ));

93+ )

94+ .into());

95+ }

98 }

96 }

99− headers.insert(parts[0].to_string(), parts[1].to_string());

100 }

97 }

101

102− // find content-length

103− let content_length = headers.get("Content-Length").unwrap().parse().unwrap();

99+ let content_length = content_length.unwrap();

104

100

101+ //TODO: reuse vector

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

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

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

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

107 let msg = String::from_utf8(content).unwrap();

104 let msg = String::from_utf8(content).unwrap();

108

105

109− // read data

110−

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

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

112

107

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

108 // 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);

115

110

116 Ok(output?)

111 Ok(output?)

117 }

112 }

118

113

119− async fn err(

114+ async fn recv_server_error(

120 err: &mut (impl AsyncBufRead + Unpin + Send),

115 err: &mut (impl AsyncBufRead + Unpin + Send),

121− ) -> core::result::Result<(), std::io::Error> {

116+ buffer: &mut String,

122− let mut line = String::new();

117+ ) -> Result<()> {

118+ buffer.truncate(0);

123− err.read_line(&mut line).await?;

119+ err.read_line(buffer).await?;

124− error!("err <- {}", line);

120+ error!("err <- {}", buffer);

125

121

126 Ok(())

122 Ok(())

127 }

123 }

128

124

129− pub async fn send_payload(&mut self, payload: Payload) -> io::Result<()> {

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

126+ //TODO: reuse string

130− match payload {

127+ let json = match payload {

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

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

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

129 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)?;

130+ serde_json::to_string(&value)?

139− self.send(json).await

140 }

131 }

141− Payload::Response(error) => {

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

142− let json = serde_json::to_string(&error)?;

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

134+ };

143− self.send(json).await

135+ self.send_string_to_server(json).await

144− }

145− }

146 }

136 }

147

137

148− pub async fn send(&mut self, request: String) -> io::Result<()> {

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

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

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

150

140

151 // send the headers

141 // send the headers

152− self.writer

142+ self.server_stdin

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

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

154 .await?;

144 .await?;

155

145

156 // send the body

146 // send the body

157− self.writer.write_all(request.as_bytes()).await?;

147+ self.server_stdin.write_all(request.as_bytes()).await?;

158

148

159− self.writer.flush().await?;

149+ self.server_stdin.flush().await?;

160

150

161 Ok(())

151 Ok(())

162 }

152 }

163

153

164− async fn recv_msg(&mut self, msg: Message) -> anyhow::Result<()> {

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

165 match msg {

155 match msg {

166− Message::Output(output) => self.recv_response(output).await?,

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

167− Message::Call(call) => {

157+ ServerMessage::Call(call) => {

168− self.incoming.send(call).unwrap();

158+ self.client_tx.send(call).unwrap();

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

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

170 }

160 }

171 };

161 };

172 Ok(())

162 Ok(())

173 }

163 }

174

164

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

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

176 let (id, result) = match output {

166 let (id, result) = match output {

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

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

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

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

@@ -200,29 +190,33 @@ impl Transport {

200 Ok(())

190 Ok(())

201 }

191 }

202

192

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();

204 loop {

196 loop {

205 tokio::select! {

197 tokio::select! {

206 // client -> server

198 // client -> server

207− msg = self.outgoing.recv() => {

199+ msg = self.client_rx.recv() => {

208− if msg.is_none() {

200+ match msg {

201+ Some(msg) => {

202+ self.send_payload_to_server(msg).await.unwrap()

203+ },

209− break;

204+ None => break

210 }

205 }

211− let msg = msg.unwrap();

212−

213− self.send_payload(msg).await.unwrap();

214 }

206 }

215− // server <- client

207+ // server -> client

216− msg = Self::recv(&mut self.reader, &mut self.headers) => {

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

217− if msg.is_err() {

209+ match msg {

210+ Ok(msg) => {

211+ self.process_server_message(msg).await.unwrap();

212+ }

213+ Err(_) => {

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

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

219− break;

215+ break;

216+ },

220 }

217 }

221− let msg = msg.unwrap();

222−

223− self.recv_msg(msg).await.unwrap();

224 }

218 }

225− _msg = Self::err(&mut self.stderr) => {}

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

226 }

220 }

227 }

221 }

228 }

222 }