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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
use futures::future::{Empty, IntoStream};
use futures::sync::mpsc;
use futures::{future, Future, Poll, Sink, Stream};
use log::error;
use tokio_codec::{FramedRead, FramedWrite};
use tokio_io::{AsyncRead, AsyncWrite};
use tower_service::Service;
use super::codec::LanguageServerCodec;
#[derive(Debug)]
pub struct Server<I, O, S = Nothing> {
stdin: I,
stdout: O,
interleave: S,
}
impl<I, O> Server<I, O, Nothing>
where
I: AsyncRead + Send + 'static,
O: AsyncWrite + Send + 'static,
{
pub fn new(stdin: I, stdout: O) -> Self {
Server {
stdin,
stdout,
interleave: Nothing::new(),
}
}
}
impl<I, O, S> Server<I, O, S>
where
I: AsyncRead + Send + 'static,
O: AsyncWrite + Send + 'static,
S: Stream<Item = String, Error = ()> + Send + 'static,
{
pub fn interleave<T>(self, stream: T) -> Server<I, O, T>
where
T: Stream<Item = String, Error = ()> + Send + 'static,
{
Server {
stdin: self.stdin,
stdout: self.stdout,
interleave: stream,
}
}
pub fn serve<T>(self, service: T) -> impl Future<Item = (), Error = ()> + Send + 'static
where
T: Service<String, Response = String> + Send + 'static,
T::Future: Send + 'static,
{
let (sender, receiver) = mpsc::channel(1);
let framed_stdin = FramedRead::new(self.stdin, LanguageServerCodec::default());
let framed_stdout = FramedWrite::new(self.stdout, LanguageServerCodec::default());
let interleave = self.interleave;
future::lazy(move || {
let printer = receiver
.select(interleave)
.map_err(|_| error!("failed to log message"))
.forward(framed_stdout.sink_map_err(|e| error!("failed to encode response: {}", e)))
.map(|_| ());
tokio_executor::spawn(printer);
framed_stdin
.map_err(|e| error!("failed to decode request: {}", e))
.fold(service, move |mut service, line| {
let sender = sender.clone();
service
.call(line)
.and_then(move |resp| sender.send(resp).map_err(|_| unreachable!()))
.then(move |_| Ok(service))
})
.map(|_| ())
})
}
}
#[doc(hidden)]
#[derive(Debug)]
pub struct Nothing(IntoStream<Empty<String, ()>>);
impl Nothing {
fn new() -> Self {
Nothing(future::empty().into_stream())
}
}
impl Stream for Nothing {
type Item = String;
type Error = ();
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
self.0.poll()
}
}