dnet-rpc 0.1.1

Remote Procedure Call over dnet transports
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

//! Return value for value requests.

use std::{
    pin::Pin,
    sync::{Arc, Mutex},
    task::{Context, Poll},
};

use futures::{channel::oneshot, future::FusedFuture, ready, Future, FutureExt};
use pin_project::{pin_project, pinned_drop};

use crate::parts::consumer::RequestSender;

use super::{Aborter, ResultSender};

/// Future returned by consumer value requests.
#[derive(Debug)]
#[pin_project(PinnedDrop)]
pub struct ValueRequest<Request, T> {
    sender: RequestSender<Request, T>,
    id: u64,
    request: Option<Request>,
    receiver: Option<oneshot::Receiver<super::Result<T>>>,
    aborter: Option<Aborter<Request, T>>,
    abort_receiver: Option<oneshot::Receiver<()>>,
}

impl<Request, T> ValueRequest<Request, T> {
    /// Create new value request.
    ///
    /// **NOTE**: This is used internally by the generated consumers.<br>
    /// You should never have to create it manually yourself.
    pub fn new(sender: RequestSender<Request, T>, id: u64, request: Request) -> Self {
        ValueRequest {
            sender,
            id,
            request: Some(request),
            receiver: None,
            aborter: None,
            abort_receiver: None,
        }
    }

    /// Request id.
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Aborter for this value request.
    pub fn aborter(&mut self) -> Aborter<Request, T> {
        let aborter = self.aborter.get_or_insert_with(|| {
            let (abort_sender, abort_receiver) = oneshot::channel();
            self.abort_receiver = Some(abort_receiver);
            Aborter {
                id: self.id,
                sender: self.sender.clone(),
                abort_sender: Arc::new(Mutex::new(Some(abort_sender))),
            }
        });
        aborter.clone()
    }
}

impl<Request, T> Future for ValueRequest<Request, T> {
    type Output = super::Result<T>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut me = self.project();
        if let Some(abort_receiver) = me.abort_receiver {
            if let Poll::Ready(result) = abort_receiver.poll_unpin(cx) {
                if result.is_ok() {
                    me.request.take();
                    me.receiver.take();
                    return Poll::Ready(Err(super::super::Error::Aborted));
                }
            }
        }

        if let Some(request) = me.request.take() {
            let (sender, mut receiver) = oneshot::channel();
            let sender = ResultSender::Value(sender);
            me.sender.send(*me.id, request, sender).map_err(|_| {
                me.receiver.take();
                super::super::Error::Shutdown
            })?;
            match receiver.poll_unpin(cx) {
                Poll::Ready(result) => {
                    me.receiver.take();
                    Poll::Ready(result.map_err(|_| super::super::Error::Dropped)?)
                }
                Poll::Pending => {
                    *me.receiver = Some(receiver);
                    Poll::Pending
                }
            }
        } else if let Some(receiver) = &mut me.receiver {
            let result = ready!(receiver.poll_unpin(cx));
            me.receiver.take();
            Poll::Ready(result.map_err(|_| super::super::Error::Dropped)?)
        } else {
            Poll::Pending
        }
    }
}

impl<Request, T> FusedFuture for ValueRequest<Request, T> {
    fn is_terminated(&self) -> bool {
        self.request.is_none() && self.receiver.is_none()
    }
}

#[pinned_drop]
impl<Request, T> PinnedDrop for ValueRequest<Request, T> {
    fn drop(self: Pin<&mut Self>) {
        if !self.is_terminated() {
            self.sender.abort(self.id);
        }
    }
}