protosocket-rpc 1.0.0

RPC using protosockets
Documentation 
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
107
108
109
110
111
112
113
114
115
116
117
118
119

use std::{
    future::Future,
    pin::Pin,
    task::{Context, Poll},
};

use crate::{
    server::{abortable::AbortableState, rpc_submitter::RpcResponse},
    Message,
};

pub struct ForwardAbortableUnaryRpc<F, T>
where
    F: Future<Output = AbortableState<crate::Result<T>>>,
    T: Message,
{
    future: F,
    id: u64,
    forward: spillway::Sender<RpcResponse<T>>,
    completed_for_drop: bool,
}
impl<F, T> Drop for ForwardAbortableUnaryRpc<F, T>
where
    F: Future<Output = AbortableState<crate::Result<T>>>,
    T: Message,
{
    fn drop(&mut self) {
        if !self.completed_for_drop {
            log::debug!("dropping unary rpc before completion: {}", self.id);
            let _ = self.forward.send(RpcResponse::Final(T::cancelled(self.id)));
        }
    }
}
impl<F, T> ForwardAbortableUnaryRpc<F, T>
where
    F: Future<Output = AbortableState<crate::Result<T>>>,
    T: Message,
{
    pub fn new(future: F, id: u64, forward: spillway::Sender<RpcResponse<T>>) -> Self {
        Self {
            future,
            id,
            forward,
            completed_for_drop: false,
        }
    }
}
impl<F, T> Future for ForwardAbortableUnaryRpc<F, T>
where
    F: Future<Output = AbortableState<crate::Result<T>>>,
    T: Message,
{
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        // SAFETY: This is a structural pin. If I'm not moved then neither is this future.
        let structurally_pinned_future =
            unsafe { self.as_mut().map_unchecked_mut(|me| &mut me.future) };
        match structurally_pinned_future.poll(context) {
            Poll::Ready(state) => {
                // SAFETY: This is a structural pin. If I'm not moved then neither is this boolean (it was an invariant for the future anyway).
                unsafe {
                    self.as_mut().get_unchecked_mut().completed_for_drop = true;
                }

                match state {
                    AbortableState::Ready(Ok(response)) => {
                        log::trace!("{} unary rpc response", self.id);
                        if let Err(_e) = self.forward.send(RpcResponse::Final(response)) {
                            log::debug!("outbound connection is closed");
                        }
                        Poll::Ready(())
                    }
                    AbortableState::Ready(Err(e)) => {
                        match e {
                            crate::Error::IoFailure(error) => {
                                log::warn!("{} io failure while servicing rpc: {error:?}", self.id);
                                let _ =
                                    self.forward.send(RpcResponse::Final(T::cancelled(self.id)));
                            }
                            crate::Error::CancelledRemotely => {
                                log::debug!("{} rpc cancelled remotely", self.id);
                                let _ =
                                    self.forward.send(RpcResponse::Final(T::cancelled(self.id)));
                            }
                            crate::Error::ConnectionIsClosed => {
                                log::debug!("{} rpc cancelled remotely", self.id);
                                let _ =
                                    self.forward.send(RpcResponse::Final(T::cancelled(self.id)));
                            }
                            crate::Error::Finished => {
                                log::debug!("{} unary rpc ended", self.id);
                                if let Err(_e) =
                                    self.forward.send(RpcResponse::Final(T::ended(self.id)))
                                {
                                    log::debug!("outbound connection is closed");
                                }
                            }
                        }
                        Poll::Ready(())
                        // cancelled
                    }
                    AbortableState::Abort => {
                        // This happens when the upstream stuff is dropped and there are no messages that can be produced. We'll send a cancellation.
                        log::debug!("{} unary rpc abort", self.id);
                        let _ = self.forward.send(RpcResponse::Final(T::cancelled(self.id)));
                        Poll::Ready(())
                    }
                    AbortableState::Aborted => {
                        log::debug!("{} unary rpc was cancelled by the client.", self.id);
                        let _ = self.forward.send(RpcResponse::Final(T::cancelled(self.id)));
                        Poll::Ready(())
                    }
                }
            }
            Poll::Pending => Poll::Pending,
        }
    }
}