lsp_max/service/client/
socket.rs

1//! Loopback connection to the language client.
2
3use std::pin::Pin;
4use std::sync::Arc;
5use std::task::{Context, Poll};
6
7use futures::channel::mpsc::Receiver;
8use futures::sink::Sink;
9use futures::stream::{FusedStream, Stream, StreamExt};
10
11use super::{ExitedError, Pending, ServerState, State};
12use crate::jsonrpc::{Request, Response};
13
14/// A loopback channel for server-to-client communication.
15///
16/// See also: `examples/transport_utilities_explained.rs` — a run-to-exit witness that
17/// demonstrates `ClientSocket`, `ExitedError`, and `Loopback` with real `assert!`s.
18#[derive(Debug)]
19pub struct ClientSocket {
20    pub(super) rx: Receiver<Request>,
21    pub(super) pending: Arc<Pending>,
22    pub(super) state: Arc<ServerState>,
23}
24
25impl ClientSocket {
26    /// Splits this `ClientSocket` into two halves capable of operating independently.
27    ///
28    /// The two halves returned implement the [`Stream`] and [`Sink`] traits, respectively.
29    ///
30    /// [`Stream`]: futures::Stream
31    /// [`Sink`]: futures::Sink
32    pub fn split(self) -> (RequestStream, ResponseSink) {
33        let ClientSocket { rx, pending, state } = self;
34        let state_ = state.clone();
35
36        (
37            RequestStream { rx, state: state_ },
38            ResponseSink { pending, state },
39        )
40    }
41}
42
43/// Yields a stream of pending server-to-client requests.
44impl Stream for ClientSocket {
45    type Item = Request;
46
47    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
48        if self.state.get() == State::Exited || self.rx.is_terminated() {
49            Poll::Ready(None)
50        } else {
51            self.rx.poll_next_unpin(cx)
52        }
53    }
54
55    fn size_hint(&self) -> (usize, Option<usize>) {
56        self.rx.size_hint()
57    }
58}
59
60impl FusedStream for ClientSocket {
61    fn is_terminated(&self) -> bool {
62        self.rx.is_terminated()
63    }
64}
65
66/// Routes client-to-server responses back to the server.
67impl Sink<Response> for ClientSocket {
68    type Error = ExitedError;
69
70    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
71        if self.state.get() == State::Exited || self.rx.is_terminated() {
72            let code = if self.state.get() == State::Exited {
73                self.state.get_exit_code()
74            } else {
75                1
76            };
77            Poll::Ready(Err(ExitedError(code)))
78        } else {
79            Poll::Ready(Ok(()))
80        }
81    }
82
83    fn start_send(self: Pin<&mut Self>, item: Response) -> Result<(), Self::Error> {
84        self.pending.insert(item);
85        Ok(())
86    }
87
88    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
89        Poll::Ready(Ok(()))
90    }
91
92    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
93        Poll::Ready(Ok(()))
94    }
95}
96
97/// Yields a stream of pending server-to-client requests.
98#[derive(Debug)]
99#[must_use = "streams do nothing unless polled"]
100pub struct RequestStream {
101    rx: Receiver<Request>,
102    state: Arc<ServerState>,
103}
104
105impl Stream for RequestStream {
106    type Item = Request;
107
108    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
109        if self.state.get() == State::Exited || self.rx.is_terminated() {
110            Poll::Ready(None)
111        } else {
112            self.rx.poll_next_unpin(cx)
113        }
114    }
115
116    fn size_hint(&self) -> (usize, Option<usize>) {
117        self.rx.size_hint()
118    }
119}
120
121impl FusedStream for RequestStream {
122    fn is_terminated(&self) -> bool {
123        self.rx.is_terminated()
124    }
125}
126
127/// Routes client-to-server responses back to the server.
128#[derive(Debug)]
129pub struct ResponseSink {
130    pending: Arc<Pending>,
131    state: Arc<ServerState>,
132}
133
134impl Sink<Response> for ResponseSink {
135    type Error = ExitedError;
136
137    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
138        if self.state.get() == State::Exited {
139            Poll::Ready(Err(ExitedError(self.state.get_exit_code())))
140        } else {
141            Poll::Ready(Ok(()))
142        }
143    }
144
145    fn start_send(self: Pin<&mut Self>, item: Response) -> Result<(), Self::Error> {
146        self.pending.insert(item);
147        Ok(())
148    }
149
150    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
151        Poll::Ready(Ok(()))
152    }
153
154    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
155        Poll::Ready(Ok(()))
156    }
157}
lsp_max/service/client/socket.rs

lsp_max/service/client/
socket.rs
