Struct Context

Source

pub struct Context { /* private fields */ }

Expand description

A socket context.

The context allows for independent and concurrent use of stateful operations on a single socket. Using contexts is an excellent way to write simpler concurrent applications, while retaining the benefits of the protocol-specific advanced processing.

Note that not all protocols allow for the creation of contexts.

§Examples

See the documentation of the Aio type for examples on how to use socket contexts.

Implementations§

Source §

impl Context

Source

pub fn new(socket: &Socket) -> Result<Context>

Creates a new socket context.

§Errors

NotSupported: The protocol does not support separate contexts or the socket was opened in raw mode.
OutOfMemory: Insufficient memory is available.

Examples found in repository ?

examples/async.rs (line 69)

62fn server(url: &str) -> Result<(), nng::Error> {
63    // Create the socket
64    let s = Socket::new(Protocol::Rep0)?;
65
66    // Create all of the worker contexts
67    let workers: Vec<_> = (0..PARALLEL)
68        .map(|_| {
69            let ctx = Context::new(&s)?;
70            let ctx_clone = ctx.clone();
71            let aio = Aio::new(move |aio, res| worker_callback(aio, &ctx_clone, res))?;
72            Ok((aio, ctx))
73        })
74        .collect::<Result<_, nng::Error>>()?;
75
76    // Only after we have the workers do we start listening.
77    s.listen(url)?;
78
79    // Now start all of the workers listening.
80    for (a, c) in &workers {
81        c.recv(a)?;
82    }
83
84    thread::sleep(Duration::from_secs(60 * 60 * 24 * 365));
85
86    Ok(())
87}

Source

pub fn send<M: Into<Message>>( &self, aio: &Aio, msg: M, ) -> Result<(), (Message, Error)>

Start a send operation on the given Aio and return immediately.

§Errors

IncorrectState: The Aio already has a running operation.

Examples found in repository ?

examples/async.rs (line 103)

90fn worker_callback(aio: Aio, ctx: &Context, res: AioResult) {
91    match res {
92        // We successfully sent the message, wait for a new one.
93        AioResult::Send(Ok(_)) => ctx.recv(&aio).unwrap(),
94
95        // We successfully received a message.
96        AioResult::Recv(Ok(m)) => {
97            let ms = u64::from_le_bytes(m[..].try_into().unwrap());
98            aio.sleep(Duration::from_millis(ms)).unwrap();
99        }
100
101        // We successfully slept.
102        AioResult::Sleep(Ok(_)) => {
103            ctx.send(&aio, Message::new()).unwrap();
104        }
105
106        // Anything else is an error and we will just panic.
107        AioResult::Send(Err((_, e))) | AioResult::Recv(Err(e)) | AioResult::Sleep(Err(e)) => {
108            panic!("Error: {}", e)
109        }
110    }
111}

Source

pub fn recv(&self, aio: &Aio) -> Result<()>

Start a receive operation using the given Aio and return immediately.

§Errors

IncorrectState: The Aio already has a running operation.

Examples found in repository ?

examples/async.rs (line 81)

62fn server(url: &str) -> Result<(), nng::Error> {
63    // Create the socket
64    let s = Socket::new(Protocol::Rep0)?;
65
66    // Create all of the worker contexts
67    let workers: Vec<_> = (0..PARALLEL)
68        .map(|_| {
69            let ctx = Context::new(&s)?;
70            let ctx_clone = ctx.clone();
71            let aio = Aio::new(move |aio, res| worker_callback(aio, &ctx_clone, res))?;
72            Ok((aio, ctx))
73        })
74        .collect::<Result<_, nng::Error>>()?;
75
76    // Only after we have the workers do we start listening.
77    s.listen(url)?;
78
79    // Now start all of the workers listening.
80    for (a, c) in &workers {
81        c.recv(a)?;
82    }
83
84    thread::sleep(Duration::from_secs(60 * 60 * 24 * 365));
85
86    Ok(())
87}
88
89/// Callback function for workers.
90fn worker_callback(aio: Aio, ctx: &Context, res: AioResult) {
91    match res {
92        // We successfully sent the message, wait for a new one.
93        AioResult::Send(Ok(_)) => ctx.recv(&aio).unwrap(),
94
95        // We successfully received a message.
96        AioResult::Recv(Ok(m)) => {
97            let ms = u64::from_le_bytes(m[..].try_into().unwrap());
98            aio.sleep(Duration::from_millis(ms)).unwrap();
99        }
100
101        // We successfully slept.
102        AioResult::Sleep(Ok(_)) => {
103            ctx.send(&aio, Message::new()).unwrap();
104        }
105
106        // Anything else is an error and we will just panic.
107        AioResult::Send(Err((_, e))) | AioResult::Recv(Err(e)) | AioResult::Sleep(Err(e)) => {
108            panic!("Error: {}", e)
109        }
110    }
111}

Source

pub fn close(&self)

Closes the context.

Messages that have been submitted for sending may be flushed or delivered, depending on the underlying transport and the linger option. Further attempts to use the context (with this or any other handle) will result in an error. Threads waiting for operations on the context when this call is executed may also return with an error.

Closing the owning socket also closes this context. Additionally, the context is closed once all handles have been dropped.