protosocket-rpc 1.0.2

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

use std::future::Future;

use crate::{
    server::{
        abortable::IdentifiableAbortable, abortion_tracker::AbortionTracker,
        forward_streaming::ForwardAbortableStreamingRpc, forward_unary::ForwardAbortableUnaryRpc,
        rpc_submitter::RpcResponse,
    },
    Message,
};

/// A request context's temporary lease to an RPC Reactor's state.
/// You want to consume your RpcResponder as quickly as possible.
#[must_use]
pub struct RpcResponder<'a, Response> {
    outbound: &'a spillway::Sender<RpcResponse<Response>>,
    aborts: &'a mut AbortionTracker,
    message_id: u64,
}
impl<'a, Response> RpcResponder<'a, Response>
where
    Response: Message,
{
    pub(crate) fn new_responder_reference(
        outbound: &'a spillway::Sender<RpcResponse<Response>>,
        aborts: &'a mut AbortionTracker,
        message_id: u64,
    ) -> Self {
        Self {
            outbound,
            aborts,
            message_id,
        }
    }

    /// Consume the responder by providing a future that will materialize the response to this request.
    pub fn unary(self, unary_rpc: impl Future<Output = Response>) -> impl Future<Output = ()> {
        let (abortable, abort) = IdentifiableAbortable::new(unary_rpc);
        self.aborts.register(self.message_id, abort);
        ForwardAbortableUnaryRpc::new(abortable, self.message_id, self.outbound.clone())
    }

    /// Consume the responder by providing a stream that will materialize the response to this request.
    pub fn stream(
        self,
        streaming_rpc: impl futures::Stream<Item = Response>,
    ) -> impl Future<Output = ()> {
        let (abortable_stream, abort) = IdentifiableAbortable::new(streaming_rpc);
        self.aborts.register(self.message_id, abort);
        ForwardAbortableStreamingRpc::new(abortable_stream, self.message_id, self.outbound.clone())
    }

    /// Consume the responder by providing an immediate response.
    ///
    /// This is the cheapest, fastest way to respond, but you must only use it when you can get a response
    /// without blocking!
    pub fn immediate(self, response: Response) {
        if self
            .outbound
            .send(RpcResponse::Untracked(response))
            .is_err()
        {
            log::debug!("outbound channel closed while sending response");
        }
    }
}