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
use std::task::Poll;
use ntex::{io::IoBoxed, io::RecvError, util::poll_fn, util::ready};
use super::cmd::Command;
use super::codec::Codec;
use super::errors::{CommandError, Error};
pub struct SimpleClient {
io: IoBoxed,
}
impl SimpleClient {
pub(crate) fn new(io: IoBoxed) -> Self {
SimpleClient { io }
}
pub async fn exec<U>(&self, cmd: U) -> Result<U::Output, CommandError>
where
U: Command,
{
self.io.encode(cmd.to_request(), &Codec)?;
poll_fn(|cx| loop {
return match ready!(self.io.poll_recv(&Codec, cx)) {
Ok(item) => Poll::Ready(U::to_output(
item.into_result().map_err(CommandError::Error)?,
)),
Err(RecvError::KeepAlive) | Err(RecvError::Stop) => {
unreachable!()
}
Err(RecvError::WriteBackpressure) => {
ready!(self.io.poll_flush(cx, false))
.map_err(|e| CommandError::Protocol(Error::PeerGone(Some(e))))?;
continue;
}
Err(RecvError::Decoder(err)) => Poll::Ready(Err(CommandError::Protocol(err))),
Err(RecvError::PeerGone(err)) => {
Poll::Ready(Err(CommandError::Protocol(Error::PeerGone(err))))
}
};
})
.await
}
}