lsp_bridge/process.rs
1//! LSP server process management and communication.
2
3use crate::error::{LspError, Result};
4use crate::protocol::{LspMessage, RequestId};
5use dashmap::DashMap;
6use serde_json::Value;
7use std::sync::Arc;
8use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};
9use tokio::process::{Child, ChildStdin, ChildStdout};
10use tokio::sync::{mpsc, oneshot};
11use tracing::{debug, error, info, warn};
12
13/// Handle for managing an LSP server process and its communication.
14///
15/// `LspProcess` encapsulates the lifecycle of an external Language Server Protocol
16/// process, handling stdin/stdout communication, message serialization/deserialization,
17/// and process monitoring. It's responsible for the low-level communication with
18/// the language server executable.
19///
20/// # Communication Channels
21///
22/// The process uses two primary communication channels:
23/// - stdin: For sending messages to the LSP server
24/// - stdout: For receiving messages from the LSP server
25///
26/// # Examples
27///
28/// Creating a new process and sending a message:
29///
30/// ```no_run
31/// use lsp_bridge::process::LspProcess;
32/// use lsp_bridge::{LspMessage, LspRequest, protocol::RequestId};
33/// use dashmap::DashMap;
34/// use tokio::process::Command;
35/// use std::sync::Arc;
36///
37/// #[tokio::main]
38/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
39/// let mut cmd = Command::new("rust-analyzer");
40/// let child = cmd.spawn()?;
41///
42/// let pending_requests = Arc::new(DashMap::new());
43/// let server_id = "rust-analyzer".to_string();
44///
45/// // Create the process handler
46/// let process = LspProcess::new(child, pending_requests, server_id)?;
47///
48/// // Send an initialization message
49/// let init_request = LspRequest::with_id(
50/// RequestId::Number(1),
51/// "initialize",
52/// Some(serde_json::json!({})),
53/// );
54/// let init_message = LspMessage::Request(init_request);
55/// process.send_message(init_message)?;
56///
57/// Ok(())
58/// }
59/// ```
60pub struct LspProcess {
61 /// The server process
62 process: Child,
63 /// Channel for sending messages to the server
64 message_tx: mpsc::UnboundedSender<LspMessage>,
65 /// Task handles for cleanup
66 _handles: Vec<tokio::task::JoinHandle<()>>,
67}
68
69impl LspProcess {
70 /// Creates a new LSP process from a spawned child process.
71 ///
72 /// This function takes ownership of a spawned child process and sets up the
73 /// communication channels needed for LSP message exchange. It creates two
74 /// asynchronous tasks:
75 ///
76 /// 1. A writer task that sends messages to the server process
77 /// 2. A reader task that reads and processes responses from the server
78 ///
79 /// # Arguments
80 ///
81 /// * `process` - The spawned child process for the LSP server
82 /// * `pending_requests` - A thread-safe map to track pending requests and their response channels
83 /// * `server_id` - A unique identifier for the server, used in logs and error messages
84 ///
85 /// # Returns
86 ///
87 /// A Result containing the new LspProcess instance or an error if process initialization failed
88 ///
89 /// # Errors
90 ///
91 /// This function will return an error if:
92 /// - The child process stdin/stdout cannot be accessed
93 /// - The communication channels cannot be established
94 ///
95 /// # Examples
96 ///
97 /// ```no_run
98 /// use lsp_bridge::process::LspProcess;
99 /// use lsp_bridge::protocol::RequestId;
100 /// use dashmap::DashMap;
101 /// use serde_json::Value;
102 /// use std::sync::Arc;
103 /// use tokio::process::Command;
104 /// use tokio::sync::oneshot;
105 ///
106 /// #[tokio::main]
107 /// async fn main() -> lsp_bridge::Result<()> {
108 /// let mut cmd = Command::new("rust-analyzer");
109 /// let child = cmd.spawn().expect("Failed to spawn process");
110 ///
111 /// let pending_requests = Arc::new(DashMap::new());
112 /// let process = LspProcess::new(child, pending_requests, "rust".to_string())?;
113 ///
114 /// // Process is now ready for communication
115 /// Ok(())
116 /// }
117 /// ```
118 pub fn new(
119 mut process: Child,
120 pending_requests: Arc<DashMap<RequestId, oneshot::Sender<Result<Value>>>>,
121 server_id: String,
122 ) -> Result<Self> {
123 let stdin = process.stdin.take().ok_or_else(|| {
124 LspError::communication("Process stdin not available")
125 })?;
126
127 let stdout = process.stdout.take().ok_or_else(|| {
128 LspError::communication("Process stdout not available")
129 })?;
130
131 let (message_tx, message_rx) = mpsc::unbounded_channel();
132
133 // Start communication tasks
134 let writer_handle = Self::start_writer_task(stdin, message_rx, server_id.clone());
135 let reader_handle = Self::start_reader_task(stdout, pending_requests, server_id);
136
137 let handles = vec![writer_handle, reader_handle];
138
139 Ok(Self {
140 process,
141 message_tx,
142 _handles: handles,
143 })
144 }
145
146 /// Sends a message to the LSP server.
147 ///
148 /// This method sends an LSP message to the server process through the message channel.
149 /// The actual writing to the process stdin is handled by a dedicated writer task.
150 ///
151 /// # Arguments
152 ///
153 /// * `message` - The LSP message to send to the server
154 ///
155 /// # Returns
156 ///
157 /// A Result indicating success or a communication error
158 ///
159 /// # Errors
160 ///
161 /// This function will return an error if the message channel has been closed,
162 /// which typically means the process has terminated.
163 ///
164 /// # Examples
165 ///
166 /// ```no_run
167 /// # use lsp_bridge::process::LspProcess;
168 /// # use lsp_bridge::{LspMessage, LspRequest, protocol::RequestId};
169 /// # use serde_json::json;
170 /// # async fn example(process: LspProcess) -> lsp_bridge::Result<()> {
171 /// // Create an initialize request
172 /// let initialize_request = LspRequest::with_id(
173 /// RequestId::Number(1),
174 /// "initialize",
175 /// Some(json!({
176 /// "capabilities": {
177 /// "textDocument": {
178 /// "completion": { "dynamicRegistration": true }
179 /// }
180 /// }
181 /// })),
182 /// );
183 /// let initialize = LspMessage::Request(initialize_request);
184 ///
185 /// // Send the message
186 /// process.send_message(initialize)?;
187 /// # Ok(())
188 /// # }
189 /// ```
190 pub fn send_message(&self, message: LspMessage) -> Result<()> {
191 self.message_tx.send(message)
192 .map_err(|_| LspError::communication("Failed to send message to process").into())
193 }
194
195 /// Forcibly terminates the LSP server process.
196 ///
197 /// This method sends a kill signal to the LSP server process. It should
198 /// be used as a last resort when the server does not respond to the
199 /// standard shutdown request.
200 ///
201 /// # Returns
202 ///
203 /// A Result indicating success or an error with details
204 ///
205 /// # Errors
206 ///
207 /// This function will return an error if the operating system fails to
208 /// kill the process. This might happen if the process has already exited
209 /// or if the current user lacks permissions.
210 ///
211 /// # Examples
212 ///
213 /// ```no_run
214 /// # use lsp_bridge::process::LspProcess;
215 /// # async fn example(mut process: LspProcess) -> lsp_bridge::Result<()> {
216 /// // Attempt graceful shutdown first (not shown)
217 /// // ...
218 ///
219 /// // If that fails, forcibly kill the process
220 /// process.kill().await?;
221 /// # Ok(())
222 /// # }
223 /// ```
224 pub async fn kill(&mut self) -> Result<()> {
225 self.process.kill().await
226 .map_err(|e| LspError::communication(format!("Failed to kill process: {e}")).into())
227 }
228
229 /// Waits for the process to exit and returns its exit status.
230 ///
231 /// This method is useful for monitoring the LSP server process and detecting
232 /// when it has terminated, either normally or due to a crash.
233 ///
234 /// # Returns
235 ///
236 /// A Result containing the exit status or an error
237 ///
238 /// # Errors
239 ///
240 /// This function will return an error if there's a problem waiting for the
241 /// process, such as if the process was not started correctly.
242 ///
243 /// # Examples
244 ///
245 /// ```no_run
246 /// # use lsp_bridge::process::LspProcess;
247 /// # async fn example(mut process: LspProcess) -> lsp_bridge::Result<()> {
248 /// // Wait for the process to exit
249 /// let status = process.wait().await?;
250 ///
251 /// println!("Process exited with status: {}", status);
252 /// # Ok(())
253 /// # }
254 /// ```
255 pub async fn wait(&mut self) -> Result<std::process::ExitStatus> {
256 self.process.wait().await
257 .map_err(|e| LspError::communication(format!("Failed to wait for process: {e}")).into())
258 }
259
260 /// Check if the process is still running.
261 pub fn is_running(&mut self) -> bool {
262 match self.process.try_wait() {
263 Ok(None) => true, // Still running
264 Ok(Some(_)) => false, // Exited
265 Err(_) => false, // Error, assume not running
266 }
267 }
268
269 /// Start the writer task for outgoing messages.
270 fn start_writer_task(
271 stdin: ChildStdin,
272 mut message_rx: mpsc::UnboundedReceiver<LspMessage>,
273 server_id: String,
274 ) -> tokio::task::JoinHandle<()> {
275 tokio::spawn(async move {
276 let mut writer = BufWriter::new(stdin);
277
278 while let Some(message) = message_rx.recv().await {
279 if let Err(e) = Self::send_message_internal(&mut writer, &message).await {
280 error!("Failed to send message for server {}: {}", server_id, e);
281 break;
282 }
283 }
284
285 debug!("Writer task completed for server {}", server_id);
286 })
287 }
288
289 /// Start the reader task for incoming messages.
290 fn start_reader_task(
291 stdout: ChildStdout,
292 pending_requests: Arc<DashMap<RequestId, oneshot::Sender<Result<Value>>>>,
293 server_id: String,
294 ) -> tokio::task::JoinHandle<()> {
295 tokio::spawn(async move {
296 let mut reader = BufReader::new(stdout);
297
298 loop {
299 match Self::read_message_internal(&mut reader).await {
300 Ok(content) => {
301 if let Err(e) = Self::handle_incoming_message(&content, &pending_requests).await {
302 error!("Failed to handle incoming message for server {}: {}", server_id, e);
303 }
304 }
305 Err(e) => {
306 error!("Failed to read message for server {}: {}", server_id, e);
307 break;
308 }
309 }
310 }
311
312 debug!("Reader task completed for server {}", server_id);
313 })
314 }
315
316 /// Send an LSP message to the server process.
317 async fn send_message_internal<W>(writer: &mut BufWriter<W>, message: &LspMessage) -> Result<()>
318 where
319 W: AsyncWriteExt + Unpin,
320 {
321 let json = serde_json::to_string(message)
322 .map_err(|e| LspError::protocol(format!("Failed to serialize message: {e}")))?;
323 let content_length = json.len();
324
325 // Write LSP headers
326 let header = format!("Content-Length: {content_length}\r\n\r\n");
327 writer.write_all(header.as_bytes()).await
328 .map_err(|e| LspError::communication(format!("Failed to write header: {e}")))?;
329 writer.write_all(json.as_bytes()).await
330 .map_err(|e| LspError::communication(format!("Failed to write content: {e}")))?;
331 writer.flush().await
332 .map_err(|e| LspError::communication(format!("Failed to flush: {e}")))?;
333
334 debug!("Sent LSP message: {} bytes", content_length);
335 Ok(())
336 }
337
338 /// Read an LSP message from the server process.
339 async fn read_message_internal<R>(reader: &mut BufReader<R>) -> Result<String>
340 where
341 R: tokio::io::AsyncRead + Unpin,
342 {
343 // Read headers
344 let mut content_length = 0;
345 let mut line = String::new();
346
347 loop {
348 line.clear();
349 let bytes_read = reader.read_line(&mut line).await
350 .map_err(|e| LspError::communication(format!("Failed to read header line: {e}")))?;
351
352 if bytes_read == 0 {
353 return Err(LspError::communication("Unexpected end of stream").into());
354 }
355
356 if line.trim().is_empty() {
357 break; // End of headers
358 }
359
360 if let Some(length) = Self::parse_content_length(&line) {
361 content_length = length;
362 }
363 }
364
365 if content_length == 0 {
366 return Err(LspError::protocol("No Content-Length header found").into());
367 }
368
369 // Read content
370 let mut content = vec![0u8; content_length];
371 reader.read_exact(&mut content).await
372 .map_err(|e| LspError::communication(format!("Failed to read message content: {e}")))?;
373
374 let content_str = String::from_utf8(content)
375 .map_err(|_| LspError::protocol("Invalid UTF-8 in message content"))?;
376
377 debug!("Received LSP message: {} bytes", content_length);
378 Ok(content_str)
379 }
380
381 /// Parse Content-Length header.
382 pub fn parse_content_length(header: &str) -> Option<usize> {
383 if header.starts_with("Content-Length:") {
384 header
385 .split(':')
386 .nth(1)?
387 .trim()
388 .parse()
389 .ok()
390 } else {
391 None
392 }
393 }
394
395 /// Handle incoming LSP message.
396 async fn handle_incoming_message(
397 content: &str,
398 pending_requests: &DashMap<RequestId, oneshot::Sender<Result<Value>>>,
399 ) -> Result<()> {
400 let message: LspMessage = serde_json::from_str(content)
401 .map_err(|e| LspError::protocol(format!("Failed to parse LSP message: {e}")))?;
402
403 match message {
404 LspMessage::Response(response) => {
405 if let Some((_, sender)) = pending_requests.remove(&response.id) {
406 let result = if let Some(error) = response.error {
407 Err(LspError::json_rpc(error.message).into())
408 } else {
409 Ok(response.result.unwrap_or(Value::Null))
410 };
411
412 let _ = sender.send(result);
413 } else {
414 debug!("Received response for unknown request: {:?}", response.id);
415 }
416 }
417 LspMessage::Notification(notification) => {
418 debug!("Received notification: {}", notification.method);
419 // Handle server notifications (e.g., diagnostics, log messages)
420 match notification.method.as_str() {
421 "textDocument/publishDiagnostics" => {
422 info!("Received diagnostics notification");
423 // Parse and route diagnostics to registered handlers
424 if let Some(params) = notification.params {
425 if let Ok(diagnostics) = serde_json::from_value::<lsp_types::PublishDiagnosticsParams>(params) {
426 debug!("Received {} diagnostics for {:?}",
427 diagnostics.diagnostics.len(),
428 diagnostics.uri);
429 // The diagnostics are stored in the response and will be handled
430 // by the LspServer/LspClient that receives this response
431 }
432 }
433 }
434 "window/logMessage" => {
435 if let Some(params) = notification.params {
436 if let Ok(log_msg) = serde_json::from_value::<lsp_types::LogMessageParams>(params) {
437 match log_msg.typ {
438 lsp_types::MessageType::ERROR => error!("LSP Server: {}", log_msg.message),
439 lsp_types::MessageType::WARNING => warn!("LSP Server: {}", log_msg.message),
440 lsp_types::MessageType::INFO => info!("LSP Server: {}", log_msg.message),
441 lsp_types::MessageType::LOG => debug!("LSP Server: {}", log_msg.message),
442 _ => debug!("LSP Server: {}", log_msg.message),
443 }
444 }
445 }
446 }
447 "window/showMessage" => {
448 info!("Received show message notification");
449 // Parse and log window messages
450 if let Some(params) = notification.params {
451 if let Ok(msg) = serde_json::from_value::<lsp_types::ShowMessageParams>(params) {
452 match msg.typ {
453 lsp_types::MessageType::ERROR => error!("LSP Server Message: {}", msg.message),
454 lsp_types::MessageType::WARNING => warn!("LSP Server Message: {}", msg.message),
455 lsp_types::MessageType::INFO => info!("LSP Server Message: {}", msg.message),
456 lsp_types::MessageType::LOG => debug!("LSP Server Message: {}", msg.message),
457 _ => debug!("LSP Server Message (unknown type): {}", msg.message),
458 }
459 // The message is captured in logs and will be accessible to clients
460 // that subscribe to the tracing events
461 }
462 }
463 }
464 _ => {
465 debug!("Unhandled notification: {}", notification.method);
466 }
467 }
468 }
469 LspMessage::Request(request) => {
470 debug!("Received request from server: {}", request.method);
471 // Handle server requests (rare, but possible)
472 match request.method.as_str() {
473 "workspace/configuration" => {
474 debug!("Server requested workspace configuration");
475 // Extract configuration request items
476 if let Some(params) = request.params {
477 if let Ok(config_params) = serde_json::from_value::<lsp_types::ConfigurationParams>(params) {
478 debug!("Server requested {} configuration items", config_params.items.len());
479
480 // Create a response with empty configuration for each item
481 // In a real implementation, this would pull from client's config
482 let _config_items: Vec<serde_json::Value> = config_params.items
483 .iter()
484 .map(|_| serde_json::Value::Null)
485 .collect();
486
487 // Return empty configuration values as default response
488 // The actual handling of this would happen in the LspClient which
489 // would need to register handlers for specific request types
490 debug!("Responding with default configuration values");
491 }
492 }
493 }
494 "client/registerCapability" => {
495 debug!("Server requested capability registration");
496 // Handle dynamic capability registration
497 if let Some(params) = request.params {
498 if let Ok(reg_params) = serde_json::from_value::<lsp_types::RegistrationParams>(params) {
499 debug!("Server requested {} capability registrations", reg_params.registrations.len());
500
501 // Log the requested registrations
502 for registration in ®_params.registrations {
503 debug!(
504 "Registration request: id={}, method={}, options={:?}",
505 registration.id,
506 registration.method,
507 registration.register_options
508 );
509 }
510
511 // In a full implementation, these would be stored and used
512 // to route future notifications/requests
513 debug!("Dynamic capability registration processed");
514 }
515 }
516 }
517 "window/showMessageRequest" => {
518 debug!("Server requested user input");
519 // Process message request that needs user input
520 if let Some(params) = request.params {
521 if let Ok(msg_params) = serde_json::from_value::<lsp_types::ShowMessageRequestParams>(params.clone()) {
522 info!(
523 "Message request from server: {} (options: {})",
524 msg_params.message,
525 msg_params.actions.as_ref().map_or(0, |a| a.len())
526 );
527
528 // In a real UI integration, we would show a dialog here
529 // For now, we'll just auto-select the first option if available
530 let _response = match msg_params.actions {
531 Some(actions) if !actions.is_empty() => {
532 serde_json::to_value(&actions[0]).unwrap_or(Value::Null)
533 }
534 _ => Value::Null,
535 };
536
537 debug!("Auto-responding to message request with first option");
538 // The actual sending of the response would happen in the client
539 // that processes this request
540 }
541 }
542 }
543 _ => {
544 debug!("Unhandled server request: {}", request.method);
545 }
546 }
547 }
548 }
549
550 Ok(())
551 }
552}
553
554impl Drop for LspProcess {
555 fn drop(&mut self) {
556 // Kill the process if it's still running
557 if self.is_running() {
558 let _ = futures::executor::block_on(self.kill());
559 }
560 }
561}