lsp_service/
server.rs

1//! A library for handling LSP servers
2//!
3//! This implements the core parts of the [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/)
4//! which is a [JSON-RPC 2.0](https://www.jsonrpc.org/specification) based protocol.
5//!
6//! A [`Server`] can be instantiated via [`Server::listen_stdio`] (or [`Server::raw_channels`] can be used for testing)
7//!
8
9use crossbeam_channel::{Receiver, Sender, bounded};
10use lsp_types::SetTraceParams;
11use serde_json::from_value;
12use std::{
13	io,
14	sync::{Arc, RwLock},
15	thread::{Builder, JoinHandle},
16};
17use tracing::{debug, level_filters::LevelFilter, trace, warn};
18
19use crate::{Client, Notification, ParseError, TracingLayer};
20
21use super::Message;
22
23mod handler;
24
25pub use handler::Handler;
26
27pub struct ThreadConnection {
28	pub sender: JoinHandle<io::Result<()>>,
29	pub receiver: JoinHandle<io::Result<()>>,
30}
31
32impl ThreadConnection {
33	pub fn is_finished(&self) -> bool {
34		self.sender.is_finished() || self.receiver.is_finished()
35	}
36}
37
38/// A set of Sender/Receiver objects for passing [`Message`s](Message) around.
39pub struct Server {
40	write_sender: Sender<Message>,
41	write_receiver: Receiver<Message>,
42	read_sender: Sender<Message>,
43	#[allow(dead_code)]
44	request_handler: JoinHandle<Result<(), io::Error>>,
45	#[allow(dead_code)]
46	read_receiver: Receiver<Message>,
47	trace_level: Arc<RwLock<LevelFilter>>,
48}
49
50impl Server {
51	/// Spawns the message handling thread. `make_handler` receives a [`Client`] for sending server-to-client messages.
52	pub fn new<T: Handler>(make_handler: impl FnOnce(Client) -> T) -> Self {
53		let (write_sender, write_receiver) = bounded::<Message>(0);
54		let (read_sender, read_receiver) = bounded::<Message>(0);
55		let handler = make_handler(Client::new(write_sender.clone()));
56
57		let handler_receiver = read_receiver.clone();
58		let handler_sender = write_sender.clone();
59		let trace_level = Arc::new(RwLock::new(LevelFilter::OFF));
60		let level_set = trace_level.clone();
61		let request_handler = Builder::new()
62			.name("LspMessageHandler".into())
63			.spawn(move || {
64				while let Ok(message) = handler_receiver.recv() {
65					trace!("LspMessageHandler -> {:#?}", &message);
66					if let Message::Notification(Notification { method, params }) = &message {
67						if method == "exit" {
68							break;
69						}
70						if method == "$/setTrace" {
71							let level = from_value::<SetTraceParams>(params.clone())
72								.map(|p| match p.value {
73									lsp_types::TraceValue::Off => LevelFilter::OFF,
74									lsp_types::TraceValue::Messages => LevelFilter::WARN,
75									lsp_types::TraceValue::Verbose => LevelFilter::TRACE,
76								})
77								.unwrap_or(LevelFilter::OFF);
78							trace!("Changing level to {:?}", level);
79							let mut level_set = level_set.write().unwrap();
80							*level_set = level;
81						}
82					}
83					let response = handler.handle(message);
84					if let Some(response) = response
85						&& let Err(e) = handler_sender.send(response)
86					{
87						warn!("Handler failed to send response {:?}", &e);
88						return Err(io::Error::other(e));
89					}
90				}
91				debug!("LspMessageHandler closing");
92				Ok(())
93			})
94			.expect("Failed to create Reader");
95		Server { write_sender, write_receiver, read_sender, read_receiver, request_handler, trace_level }
96	}
97
98	pub fn tracer(&self) -> TracingLayer {
99		TracingLayer::new(self.trace_level.clone(), self.write_sender.clone())
100	}
101
102	pub fn listen_stdio(&self) -> Result<ThreadConnection, io::Error> {
103		let write_receiver = self.write_receiver.clone();
104		let writer = Builder::new().name("LspWriter".into()).spawn(move || {
105			let mut stdout = io::stdout().lock();
106			while let Ok(message) = write_receiver.recv() {
107				trace!("{:#?}", message);
108				message.write(&mut stdout)?;
109			}
110			Ok(())
111		})?;
112		let read_sender = self.read_sender.clone();
113		let reader = Builder::new()
114			.name("LspReader".into())
115			.spawn(move || read_messages(&mut io::stdin().lock(), &read_sender))?;
116		Ok(ThreadConnection { sender: reader, receiver: writer })
117	}
118
119	/// Returns the channels a transport would use: send client messages into the first, receive server messages from
120	/// the second. Drives the server in-process without stdio.
121	pub fn raw_channels(&self) -> (Sender<Message>, Receiver<Message>) {
122		(self.read_sender.clone(), self.write_receiver.clone())
123	}
124}
125
126/// Forwards each message read from `reader` to `sender` until the input ends or the client sends `exit`. A message
127/// whose body is not a JSON-RPC message is skipped, as its framing still says where the next one starts.
128fn read_messages(reader: &mut impl io::BufRead, sender: &Sender<Message>) -> io::Result<()> {
129	loop {
130		match Message::read(reader) {
131			Ok(Some(message)) => {
132				let exit = message.is_exit_notification();
133				sender.send(message).map_err(io::Error::other)?;
134				if exit {
135					return Ok(());
136				}
137			}
138			Ok(None) => return Ok(()),
139			Err(ParseError::Body(error)) => warn!("Skipping unparseable message: {error}"),
140			Err(error) => return Err(error.into()),
141		}
142	}
143}
144
145#[cfg(test)]
146mod tests {
147	use std::sync::atomic::{AtomicBool, Ordering};
148
149	use crate::{ErrorCode, Notification, Request, Response, ResponseError};
150
151	use super::*;
152	use lsp_types::{
153		InitializeParams, InitializeResult,
154		request::{GotoDeclaration, Initialize, Request as RequestTrait},
155	};
156	use serde_json::{Value, json, to_value};
157	use tracing::level_filters::LevelFilter;
158	use tracing_subscriber::{Layer, fmt, layer::SubscriberExt, registry, util::SubscriberInitExt};
159
160	#[test]
161	fn smoke_test() {
162		let stderr_log = fmt::layer().with_writer(io::stderr).with_filter(LevelFilter::TRACE);
163		struct TestHandler {
164			initialized: AtomicBool,
165		}
166		impl Handler for TestHandler {
167			fn initialized(&self) -> bool {
168				self.initialized.load(Ordering::SeqCst)
169			}
170			fn initialize(&self, _req: InitializeParams) -> Result<InitializeResult, ResponseError> {
171				self.initialized.swap(true, Ordering::SeqCst);
172				Ok(InitializeResult { ..Default::default() })
173			}
174		}
175
176		let server = Server::new(|_| TestHandler { initialized: AtomicBool::new(false) });
177		registry().with(stderr_log).with(server.tracer()).init();
178		let (sender, receiver) = server.raw_channels();
179		sender
180			.send(Message::Request(Request {
181				id: 1.into(),
182				method: Initialize::METHOD.into(),
183				params: to_value(InitializeParams { ..Default::default() }).unwrap(),
184			}))
185			.unwrap();
186		assert_eq!(receiver.recv(), Ok(Message::Response(Response::Ok(1.into(), json!({"capabilities": {}})))));
187		sender
188			.send(Message::Request(Request {
189				id: 1.into(),
190				method: GotoDeclaration::METHOD.into(),
191				params: json!({
192					"textDocument": {
193						"uri": "foo/bar",
194					},
195					"position": {
196						"line": 1,
197						"character": 1
198					}
199				}),
200			}))
201			.unwrap();
202		assert!(matches!(
203			receiver.recv(),
204			Ok(Message::Response(Response::Err(id, ErrorCode::MethodNotFound, _, Value::Null))) if id == 1.into()
205		));
206		sender.send(Message::Notification(Notification { method: "exit".into(), params: Value::Null })).unwrap();
207	}
208
209	#[test]
210	fn reader_skips_unparseable_bodies_and_stops_at_exit() {
211		let input = concat!(
212			"Content-Length: 11\r\n\r\n{\"foo\":\"x\"}",
213			"Content-Length: 29\r\n\r\n{\"method\":\"exit\",\"params\":[]}",
214			"Content-Length: 29\r\n\r\n{\"method\":\"next\",\"params\":[]}",
215		);
216		let (sender, receiver) = crossbeam_channel::unbounded();
217		read_messages(&mut input.as_bytes(), &sender).unwrap();
218		assert_eq!(
219			receiver.try_iter().collect::<Vec<_>>(),
220			vec![Message::Notification(Notification { method: "exit".into(), params: json!([]) })]
221		);
222	}
223}
lsp_service/server.rs

lsp_service/
server.rs
