mod msg;
mod stdio;
mod error;
mod socket;
mod req_queue;
use std::net::{TcpStream, ToSocketAddrs};
use crossbeam_channel::{Receiver, Sender};
pub use crate::{
error::ProtocolError,
msg::{ErrorCode, Message, Notification, Request, RequestId, Response, ResponseError},
req_queue::{Incoming, Outgoing, ReqQueue},
stdio::IoThreads,
};
pub struct Connection {
pub sender: Sender<Message>,
pub receiver: Receiver<Message>,
}
impl Connection {
pub fn stdio() -> (Connection, IoThreads) {
let (sender, receiver, io_threads) = stdio::stdio_transport();
(Connection { sender, receiver }, io_threads)
}
pub fn socket<A: ToSocketAddrs>(addr: A) -> (Connection, IoThreads) {
let stream = TcpStream::connect(addr).expect("Couldn't connect to the server...");
let (sender, receiver, io_threads) = socket::socket_transport(stream);
(Connection { sender, receiver }, io_threads)
}
pub fn memory() -> (Connection, Connection) {
let (s1, r1) = crossbeam_channel::unbounded();
let (s2, r2) = crossbeam_channel::unbounded();
(Connection { sender: s1, receiver: r2 }, Connection { sender: s2, receiver: r1 })
}
pub fn initialize_start(&self) -> Result<(RequestId, serde_json::Value), ProtocolError> {
loop {
match self.receiver.recv() {
Ok(Message::Request(req)) => {
if req.is_initialize() {
return Ok((req.id, req.params));
} else {
let resp = Response::new_err(
req.id.clone(),
ErrorCode::ServerNotInitialized as i32,
format!("expected initialize request, got {:?}", req),
);
self.sender.send(resp.into()).unwrap();
}
}
msg => {
return Err(ProtocolError(format!(
"expected initialize request, got {:?}",
msg
)))
}
};
}
}
pub fn initialize_finish(
&self,
initialize_id: RequestId,
initialize_result: serde_json::Value,
) -> Result<(), ProtocolError> {
let resp = Response::new_ok(initialize_id, initialize_result);
self.sender.send(resp.into()).unwrap();
match &self.receiver.recv() {
Ok(Message::Notification(n)) if n.is_initialized() => (),
m => {
return Err(ProtocolError(format!(
"expected initialized notification, got {:?}",
m
)))
}
}
Ok(())
}
pub fn initialize(
&self,
server_capabilities: serde_json::Value,
) -> Result<serde_json::Value, ProtocolError> {
let (id, params) = self.initialize_start()?;
let initialize_data = serde_json::json!({
"capabilities": server_capabilities,
});
self.initialize_finish(id, initialize_data)?;
Ok(params)
}
pub fn handle_shutdown(&self, req: &Request) -> Result<bool, ProtocolError> {
if !req.is_shutdown() {
return Ok(false);
}
let resp = Response::new_ok(req.id.clone(), ());
let _ = self.sender.send(resp.into());
match &self.receiver.recv_timeout(std::time::Duration::from_secs(30)) {
Ok(Message::Notification(n)) if n.is_exit() => (),
m => return Err(ProtocolError(format!("unexpected message during shutdown: {:?}", m))),
}
Ok(true)
}
}