lsp_server/
socket.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
use std::{
    io::{self, BufReader},
    net::TcpStream,
    thread,
};

use crossbeam_channel::{Receiver, Sender, bounded};

use crate::{
    Message,
    stdio::{IoThreads, make_io_threads},
};

pub(crate) fn socket_transport(
    stream: TcpStream,
) -> (Sender<Message>, Receiver<Message>, IoThreads) {
    let (reader_receiver, reader) = make_reader(stream.try_clone().unwrap());
    let (writer_sender, writer, messages_to_drop) = make_write(stream);
    let dropper = std::thread::spawn(move || {
        messages_to_drop.into_iter().for_each(drop);
    });
    let io_threads = make_io_threads(reader, writer, dropper);
    (writer_sender, reader_receiver, io_threads)
}

fn make_reader(stream: TcpStream) -> (Receiver<Message>, thread::JoinHandle<io::Result<()>>) {
    let (reader_sender, reader_receiver) = bounded::<Message>(0);
    let reader = thread::spawn(move || {
        let mut buf_read = BufReader::new(stream);
        while let Some(msg) = Message::read(&mut buf_read).unwrap() {
            let is_exit = matches!(&msg, Message::Notification(n) if n.is_exit());
            reader_sender.send(msg).unwrap();
            if is_exit {
                break;
            }
        }
        Ok(())
    });
    (reader_receiver, reader)
}

fn make_write(
    mut stream: TcpStream,
) -> (Sender<Message>, thread::JoinHandle<io::Result<()>>, Receiver<Message>) {
    let (writer_sender, writer_receiver) = bounded::<Message>(0);
    let (drop_sender, drop_receiver) = bounded::<Message>(0);
    let writer = thread::spawn(move || {
        writer_receiver
            .into_iter()
            .try_for_each(|it| {
                let result = it.write(&mut stream);
                let _ = drop_sender.send(it);
                result
            })
            .unwrap();
        Ok(())
    });
    (writer_sender, writer, drop_receiver)
}