runng 0.3.2

High-level wrapper around nng (Nanomsg-Next-Generation) aka Nanomsg2
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! Async request/reply

use super::*;
use crate::ctx::NngCtx;
use log::{debug, info};

#[derive(Debug, PartialEq)]
enum RequestState {
    Ready,
    Sending,
    Receiving,
}

#[derive(Debug)]
struct RequestContextAioArg {
    aio: NngAio,
    ctx: NngCtx,
    sender: Option<oneshot::Sender<Result<NngMsg>>>,
    socket: NngSocket,
    state: RequestState,
}

impl RequestContextAioArg {
    pub fn new(socket: NngSocket) -> Result<AioArg<Self>> {
        let ctx = NngCtx::new(socket.clone())?;
        NngAio::create(
            |aio| Self {
                aio,
                ctx,
                sender: None,
                socket,
                state: RequestState::Ready,
            },
            request_callback,
        )
    }
    pub fn send(&mut self, msg: NngMsg, sender: oneshot::Sender<Result<NngMsg>>) {
        if self.state != RequestState::Ready {
            panic!();
        }
        self.sender = Some(sender);
        unsafe {
            let aio = self.aio.nng_aio();
            let ctx = self.ctx.ctx();
            self.state = RequestState::Sending;

            // Nng assumes ownership of the message
            let msg = msg.take();
            nng_aio_set_msg(aio, msg);
            nng_ctx_send(ctx, aio);
        }
    }
}

impl Aio for RequestContextAioArg {
    fn aio(&self) -> &NngAio {
        &self.aio
    }
    fn aio_mut(&mut self) -> &mut NngAio {
        &mut self.aio
    }
}

/// Async request context for request/reply pattern.
#[derive(Debug)]
pub struct RequestAsyncHandle {
    aio_arg: AioArg<RequestContextAioArg>,
}

impl AsyncContext for RequestAsyncHandle {
    fn new(socket: NngSocket) -> Result<Self> {
        let aio_arg = RequestContextAioArg::new(socket)?;
        let ctx = Self { aio_arg };
        Ok(ctx)
    }
}

/// Trait for async contexts that can send a request and receive a reply.
pub trait AsyncRequest {
    /// Asynchronously send a request and return a future for the reply.
    fn send(&mut self, msg: NngMsg) -> oneshot::Receiver<Result<NngMsg>>;
}

impl AsyncRequest for RequestAsyncHandle {
    fn send(&mut self, msg: NngMsg) -> oneshot::Receiver<Result<NngMsg>> {
        let (sender, receiver) = oneshot::channel::<Result<NngMsg>>();
        self.aio_arg.send(msg, sender);
        receiver
    }
}

unsafe extern "C" fn request_callback(arg: AioArgPtr) {
    let ctx = &mut *(arg as *mut RequestContextAioArg);
    let aionng = ctx.aio.nng_aio();
    let ctxnng = ctx.ctx.ctx();
    trace!("callback Request:{:?}", ctx.state);
    match ctx.state {
        RequestState::Ready => panic!(),
        RequestState::Sending => {
            let res = nng_int_to_result(nng_aio_result(aionng));
            match res {
                Err(res) => {
                    // Nng requries we resume ownership of the message
                    let _ = NngMsg::from_raw(nng_aio_get_msg(aionng));

                    ctx.state = RequestState::Ready;
                    let sender = ctx.sender.take().unwrap();
                    sender.send(Err(res)).unwrap();
                }
                Ok(()) => {
                    ctx.state = RequestState::Receiving;
                    nng_ctx_recv(ctxnng, aionng);
                }
            }
        }
        RequestState::Receiving => {
            let sender = ctx.sender.take().unwrap();
            let res = nng_int_to_result(nng_aio_result(aionng));
            match res {
                Err(res) => {
                    ctx.state = RequestState::Ready;
                    let res = sender.send(Err(res));
                    if let Err(res) = res {
                        debug!("Receive failed to send error: {:?}", res);
                    }
                }
                Ok(()) => {
                    let msg = NngMsg::from_raw(nng_aio_get_msg(aionng));

                    ctx.state = RequestState::Ready;
                    let res = sender.send(Ok(msg));
                    if let Err(msg) = res {
                        info!("Dropping request: {:?}", msg);
                    }
                }
            }
        }
    }
}