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
140
141
142
143
144
145
146
147
148
149
150
151
152

//! Async push/pull ("pipeline")

use super::*;
use crate::protocol::*;

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

#[derive(Debug)]
struct PullContextAioArg {
    aio: NngAio,
    state: PullState,
    sender: mpsc::Sender<Result<NngMsg>>,
    socket: NngSocket,
}

impl PullContextAioArg {
    pub fn new(socket: NngSocket, sender: mpsc::Sender<Result<NngMsg>>) -> Result<AioArg<Self>> {
        NngAio::create(
            |aio| Self {
                aio,
                state: PullState::Ready,
                sender,
                socket,
            },
            pull_callback,
        )
    }

    pub(crate) fn start_receive(&mut self) {
        self.state = PullState::Receiving;
        unsafe {
            nng_recv_aio(self.socket.nng_socket(), self.aio.nng_aio());
        }
    }
}

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

/// Asynchronous context for pull socket that can receive a stream of messages.
#[derive(Debug)]
pub struct PullAsyncStream {
    aio_arg: AioArg<PullContextAioArg>,
    receiver: Option<mpsc::Receiver<Result<NngMsg>>>,
}

impl AsyncStreamContext for PullAsyncStream {
    fn new(socket: NngSocket, buffer: usize) -> Result<Self> {
        let (sender, receiver) = mpsc::channel::<Result<NngMsg>>(buffer);
        let aio_arg = PullContextAioArg::new(socket, sender)?;
        let receiver = Some(receiver);
        Ok(Self { aio_arg, receiver })
    }
}

/// Trait for asynchronous contexts that can receive a stream of messages.
pub trait AsyncPull {
    /// Asynchronously receive a stream of messages.
    fn receive(&mut self) -> Option<mpsc::Receiver<Result<NngMsg>>>;
}

impl AsyncPull for PullAsyncStream {
    fn receive(&mut self) -> Option<mpsc::Receiver<Result<NngMsg>>> {
        let receiver = self.receiver.take();
        if receiver.is_some() {
            self.aio_arg.start_receive();
        }
        receiver
    }
}

unsafe extern "C" fn pull_callback(arg: AioArgPtr) {
    let ctx = &mut *(arg as *mut PullContextAioArg);
    trace!("pull_callback::{:?}", ctx.state);
    match ctx.state {
        PullState::Ready => panic!(),
        PullState::Receiving => {
            let aio = ctx.aio.nng_aio();
            let aio_res = nng_aio_result(aio);
            let res = nng_int_to_result(aio_res);
            match res {
                Err(res) => {
                    match res {
                        // nng_aio_close() calls nng_aio_stop which nng_aio_abort(NNG_ECANCELED) and waits.
                        // If we call start_receive() it will fail with ECANCELED and we infinite loop...
                        Error::Errno(NngErrno::ECLOSED) | Error::Errno(NngErrno::ECANCELED) => {
                            debug!("pull_callback {:?}", res);
                        }
                        _ => {
                            trace!("pull_callback::Err({:?})", res);
                            ctx.start_receive();
                        }
                    }
                    try_signal_complete(&mut ctx.sender, Err(res));
                }
                Ok(()) => {
                    let msg = NngMsg::from_raw(nng_aio_get_msg(aio));
                    // Make sure to reset state before signaling completion.  Otherwise
                    // have race-condition where receiver can receive None promise
                    ctx.start_receive();
                    try_signal_complete(&mut ctx.sender, Ok(msg));
                }
            }
        }
    }
}

/// Asynchronous context for subscribe socket.
#[derive(Debug)]
pub struct SubscribeAsyncStream {
    ctx: PullAsyncStream,
}

impl AsyncPull for SubscribeAsyncStream {
    fn receive(&mut self) -> Option<mpsc::Receiver<Result<NngMsg>>> {
        self.ctx.receive()
    }
}

impl AsyncStreamContext for SubscribeAsyncStream {
    /// Create an asynchronous context using the specified socket.
    fn new(socket: NngSocket, buffer: usize) -> Result<Self> {
        let ctx = PullAsyncStream::new(socket, buffer)?;
        let ctx = Self { ctx };
        Ok(ctx)
    }
}

impl InternalSocket for SubscribeAsyncStream {
    fn socket(&self) -> &NngSocket {
        &self.ctx.aio_arg.socket
    }
}

impl Subscribe for SubscribeAsyncStream {
    fn subscribe(&self, topic: &[u8]) -> Result<()> {
        unsafe { subscribe(self.socket().nng_socket(), topic) }
    }
    fn unsubscribe(&self, topic: &[u8]) -> Result<()> {
        unsafe { unsubscribe(self.socket().nng_socket(), topic) }
    }
}