Struct Socket

pub struct Socket { /* private fields */ }

An NNG socket.

All communication between application and remote Scalability Protocol peers is done through sockets. A given socket can have multiple dialers, listeners, and pipes, and may be connected to multiple transports at the same time. However, a given socket will have exactly one protocol associated with it and is responsible for any state machines or other application-specific logic.

See the NNG documentation for more information.

Implementations

impl Socket

pub fn new(t: Protocol) -> Result<Socket>

Creates a new socket which uses the specified protocol.

Errors

• NotSupported: Protocol is not enabled.
• OutOfMemory: Insufficient memory available.

Examples found in repository

examples/pair.rs (line 31)

fn node0(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pair0)?;
    s.listen(url)?;

    send_recv(&s, "NODE0")
}

/// The dialing node.
fn node1(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pair0)?;
    s.dial(url)?;

    send_recv(&s, "NODE1")
}

examples/pipeline.rs (line 29)

fn pull(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pull0)?;
    s.listen(url)?;

    loop {
        let msg = s.recv()?;
        let arg = str::from_utf8(&msg).expect("message has invalid UTF-8");

        println!("PULL: RECEIVED \"{}\"", arg);
    }
}

/// Push socket.
fn push(url: &str, arg: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Push0)?;
    s.dial(url)?;

    println!("PUSH: SENDING \"{}\"", arg);
    s.send(arg.as_bytes())?;

    // Wait for messages to flush before shutting down.
    thread::sleep(Duration::from_secs(1));
    Ok(())
}

examples/reqrep.rs (line 35)

fn request(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    println!("REQUEST: SENDING DATE REQUEST");
    s.send(DATE_REQUEST.to_le_bytes())?;

    println!("REQUEST: WAITING FOR RESPONSE");
    let msg = s.recv()?;
    let epoch = u64::from_le_bytes(msg[..].try_into().unwrap());

    println!("REQUEST: UNIX EPOCH WAS {} SECONDS AGO", epoch);

    Ok(())
}

/// Run the reply portion of the program.
fn reply(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Rep0)?;
    s.listen(url)?;

    loop {
        println!("REPLY: WAITING FOR COMMAND");
        let mut msg = s.recv()?;

        let cmd = u64::from_le_bytes(msg[..].try_into().unwrap());
        if cmd != DATE_REQUEST {
            println!("REPLY: UNKNOWN COMMAND");
            continue;
        }

        println!("REPLY: RECEIVED DATE REQUEST");
        let rep = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("current system time is before Unix epoch")
            .as_secs();

        msg.clear();
        msg.push_back(&rep.to_le_bytes());

        println!("REPLY: SENDING {}", rep);
        s.send(msg)?;
    }
}

examples/async.rs (line 44)

fn client(url: &str, ms: u64) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    let start = Instant::now();
    s.send(ms.to_le_bytes())?;
    s.recv()?;

    let dur = Instant::now().duration_since(start);
    let subsecs: u64 = dur.subsec_millis().into();
    println!(
        "Request took {} milliseconds",
        dur.as_secs() * 1000 + subsecs
    );

    Ok(())
}

/// Run the server portion of the program.
fn server(url: &str) -> Result<(), nng::Error> {
    // Create the socket
    let s = Socket::new(Protocol::Rep0)?;

    // Create all of the worker contexts
    let workers: Vec<_> = (0..PARALLEL)
        .map(|_| {
            let ctx = Context::new(&s)?;
            let ctx_clone = ctx.clone();
            let aio = Aio::new(move |aio, res| worker_callback(aio, &ctx_clone, res))?;
            Ok((aio, ctx))
        })
        .collect::<Result<_, nng::Error>>()?;

    // Only after we have the workers do we start listening.
    s.listen(url)?;

    // Now start all of the workers listening.
    for (a, c) in &workers {
        c.recv(a)?;
    }

    thread::sleep(Duration::from_secs(60 * 60 * 24 * 365));

    Ok(())
}

examples/bus.rs (line 26)

fn node(name: &str, listen: &str, dial: &[String]) -> Result<(), Error> {
    let s = Socket::new(Protocol::Bus0)?;
    s.listen(listen)?;

    // Give time for peers to bind.
    thread::sleep(Duration::from_secs(1));
    for peer in dial {
        s.dial(peer)?;
    }

    // SEND
    println!("{0}: SENDING \"{0}\" ONTO BUS", name);
    s.send(name.as_bytes())?;

    // RECV
    loop {
        let msg = s.recv()?;
        let peer = str::from_utf8(&msg).expect("invalid UTF-8");

        println!("{}: RECEIVED \"{}\" FROM BUS", name, peer);
    }
}

examples/survey.rs (line 31)

fn surveyor(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Surveyor0)?;
    s.listen(url)?;

    loop {
        println!("SURVEYOR: SENDING DATE SURVEY REQUEST");
        s.send(DATE.as_bytes())?;

        loop {
            let msg = match s.recv() {
                Ok(m) => m,
                Err(Error::TimedOut) => break,
                Err(e) => return Err(e),
            };

            let date = u64::from_le_bytes(msg[..].try_into().unwrap());
            println!("SURVEYOR: RECEIVED \"{}\" SURVEY RESPONSE", date);
        }

        println!("SURVEYOR SURVEY COMPLETE");
    }
}

/// Respondent socket.
fn respondent(url: &str, name: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Respondent0)?;
    s.dial(url)?;

    loop {
        let mut msg = s.recv()?;

        let survey = str::from_utf8(&msg).expect("invalid UTF-8");
        println!(
            "RESPONDENT ({}): RECEIVED \"{}\" SURVEY REQUEST",
            name, survey
        );

        // Reuse the message to avoid allocation.
        msg.clear();
        let date = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("system time is before Unix epoch")
            .as_secs();

        msg.push_back(&date.to_le_bytes());

        println!("RESPONDENT ({}): SENDING \"{}\"", name, date);
        s.send(msg)?;
    }
}

Additional examples can be found in:

• examples/pubsub.rs

pub fn dial(&self, url: &str) -> Result<()>

Initiates a remote connection to a listener.

When the connection is closed, the underlying Dialer will attempt to re-establish the connection.

The first attempt to connect to the address indicated by the provided url is done synchronously, including any necessary name resolution. As a result, a failure, such as if the connection is refused, will be returned immediately and no further action will be taken.

If the connection was closed for a synchronously dialed connection, the dialer will still attempt to redial asynchronously.

Because the dialer is started immediately, it is generally not possible to apply extra configuration. If that is needed, or if one wishes to close the dialer before the socket, applications should consider using the Dialer type directly.

See the NNG documentation for more information.

Errors

• AddressInvalid: An invalid url was specified.
• Closed: The socket is not open.
• ConnectionRefused: The remote peer refused the connection.
• ConnectionReset: The remote peer reset the connection.
• DestUnreachable: The remote address is not reachable.
• OutOfMemory: Insufficient memory is available.
• PeerAuth: Authentication or authorization failure.
• Protocol: A protocol error occurred.

Examples found in repository

examples/pair.rs (line 40)

fn node1(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pair0)?;
    s.dial(url)?;

    send_recv(&s, "NODE1")
}

examples/pipeline.rs (line 43)

fn push(url: &str, arg: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Push0)?;
    s.dial(url)?;

    println!("PUSH: SENDING \"{}\"", arg);
    s.send(arg.as_bytes())?;

    // Wait for messages to flush before shutting down.
    thread::sleep(Duration::from_secs(1));
    Ok(())
}

examples/reqrep.rs (line 36)

fn request(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    println!("REQUEST: SENDING DATE REQUEST");
    s.send(DATE_REQUEST.to_le_bytes())?;

    println!("REQUEST: WAITING FOR RESPONSE");
    let msg = s.recv()?;
    let epoch = u64::from_le_bytes(msg[..].try_into().unwrap());

    println!("REQUEST: UNIX EPOCH WAS {} SECONDS AGO", epoch);

    Ok(())
}

examples/async.rs (line 45)

fn client(url: &str, ms: u64) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    let start = Instant::now();
    s.send(ms.to_le_bytes())?;
    s.recv()?;

    let dur = Instant::now().duration_since(start);
    let subsecs: u64 = dur.subsec_millis().into();
    println!(
        "Request took {} milliseconds",
        dur.as_secs() * 1000 + subsecs
    );

    Ok(())
}

examples/pubsub.rs (line 66)

fn subscriber(url: &str) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Sub0)?;
    s.dial(url)?;

    println!("SUBSCRIBER: SUBSCRIBING TO ALL TOPICS");
    let all_topics = vec![];
    s.set_opt::<Subscribe>(all_topics)?;

    loop {
        let msg = s.recv()?;
        let subs = usize::from_le_bytes(msg[..].try_into().unwrap());
        println!("SUBSCRIBER: THERE ARE {} SUBSCRIBERS", subs);
    }
}

examples/bus.rs (line 32)

fn node(name: &str, listen: &str, dial: &[String]) -> Result<(), Error> {
    let s = Socket::new(Protocol::Bus0)?;
    s.listen(listen)?;

    // Give time for peers to bind.
    thread::sleep(Duration::from_secs(1));
    for peer in dial {
        s.dial(peer)?;
    }

    // SEND
    println!("{0}: SENDING \"{0}\" ONTO BUS", name);
    s.send(name.as_bytes())?;

    // RECV
    loop {
        let msg = s.recv()?;
        let peer = str::from_utf8(&msg).expect("invalid UTF-8");

        println!("{}: RECEIVED \"{}\" FROM BUS", name, peer);
    }
}

Additional examples can be found in:

• examples/survey.rs

pub fn listen(&self, url: &str) -> Result<()>

Initiates and starts a listener on the specified address.

Listeners are used to accept connections initiated by remote dialers. Unlike a dialer, listeners generally can have many connections open concurrently.

The act of “binding” to the address indicated by url is done synchronously, including any necessary name resolution. As a result, a failure, such as if the address is already in use, will be returned immediately.

Because the listener is started immediately, it is generally not possible to apply extra configuration. If that is needed, or if one wishes to close the dialer before the socket, applications should consider using the Listener type directly.

See the NNG documentation for more information.

Errors

• AddressInUse: The address specified by url is already in use.
• AddressInvalid: An invalid url was specified.
• Closed: The socket is not open.
• OutOfMemory: Insufficient memory is available.

Examples found in repository

examples/pair.rs (line 32)

fn node0(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pair0)?;
    s.listen(url)?;

    send_recv(&s, "NODE0")
}

examples/pipeline.rs (line 30)

fn pull(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pull0)?;
    s.listen(url)?;

    loop {
        let msg = s.recv()?;
        let arg = str::from_utf8(&msg).expect("message has invalid UTF-8");

        println!("PULL: RECEIVED \"{}\"", arg);
    }
}

examples/bus.rs (line 27)

fn node(name: &str, listen: &str, dial: &[String]) -> Result<(), Error> {
    let s = Socket::new(Protocol::Bus0)?;
    s.listen(listen)?;

    // Give time for peers to bind.
    thread::sleep(Duration::from_secs(1));
    for peer in dial {
        s.dial(peer)?;
    }

    // SEND
    println!("{0}: SENDING \"{0}\" ONTO BUS", name);
    s.send(name.as_bytes())?;

    // RECV
    loop {
        let msg = s.recv()?;
        let peer = str::from_utf8(&msg).expect("invalid UTF-8");

        println!("{}: RECEIVED \"{}\" FROM BUS", name, peer);
    }
}

examples/survey.rs (line 32)

fn surveyor(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Surveyor0)?;
    s.listen(url)?;

    loop {
        println!("SURVEYOR: SENDING DATE SURVEY REQUEST");
        s.send(DATE.as_bytes())?;

        loop {
            let msg = match s.recv() {
                Ok(m) => m,
                Err(Error::TimedOut) => break,
                Err(e) => return Err(e),
            };

            let date = u64::from_le_bytes(msg[..].try_into().unwrap());
            println!("SURVEYOR: RECEIVED \"{}\" SURVEY RESPONSE", date);
        }

        println!("SURVEYOR SURVEY COMPLETE");
    }
}

examples/async.rs (line 77)

fn server(url: &str) -> Result<(), nng::Error> {
    // Create the socket
    let s = Socket::new(Protocol::Rep0)?;

    // Create all of the worker contexts
    let workers: Vec<_> = (0..PARALLEL)
        .map(|_| {
            let ctx = Context::new(&s)?;
            let ctx_clone = ctx.clone();
            let aio = Aio::new(move |aio, res| worker_callback(aio, &ctx_clone, res))?;
            Ok((aio, ctx))
        })
        .collect::<Result<_, nng::Error>>()?;

    // Only after we have the workers do we start listening.
    s.listen(url)?;

    // Now start all of the workers listening.
    for (a, c) in &workers {
        c.recv(a)?;
    }

    thread::sleep(Duration::from_secs(60 * 60 * 24 * 365));

    Ok(())
}

examples/reqrep.rs (line 53)

fn reply(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Rep0)?;
    s.listen(url)?;

    loop {
        println!("REPLY: WAITING FOR COMMAND");
        let mut msg = s.recv()?;

        let cmd = u64::from_le_bytes(msg[..].try_into().unwrap());
        if cmd != DATE_REQUEST {
            println!("REPLY: UNKNOWN COMMAND");
            continue;
        }

        println!("REPLY: RECEIVED DATE REQUEST");
        let rep = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("current system time is before Unix epoch")
            .as_secs();

        msg.clear();
        msg.push_back(&rep.to_le_bytes());

        println!("REPLY: SENDING {}", rep);
        s.send(msg)?;
    }
}

Additional examples can be found in:

• examples/pubsub.rs

pub fn dial_async(&self, url: &str) -> Result<()>

Asynchronously initiates a remote connection to a listener.

When the connection is closed, the underlying Dialer will attempt to re-establish the connection. It will also periodically retry a connection automatically if an attempt to connect asynchronously fails.

Because the dialer is started immediately, it is generally not possible to apply extra configuration. If that is needed, or if one wishes to close the dialer before the socket, applications should consider using the Dialer type directly.

See the NNG documentation for more information.

Errors

• AddressInvalid: An invalid url was specified.
• Closed: The socket is not open.
• ConnectionRefused: The remote peer refused the connection.
• ConnectionReset: The remote peer reset the connection.
• DestUnreachable: The remote address is not reachable.
• OutOfMemory: Insufficient memory is available.
• PeerAuth: Authentication or authorization failure.
• Protocol: A protocol error occurred.

pub fn recv(&self) -> Result<Message>

Receives a message from the socket.

The semantics of what receiving a message means vary from protocol to protocol, so examination of the protocol documentation is encouraged. For example, with a req socket a message may only be received after a request has been sent. Furthermore, some protocols may not support receiving data at all, such as pub.

Errors

• Closed: The socket is not open.
• IncorrectState: The socket cannot receive data in this state.
• NotSupported: The protocol does not support receiving.
• OutOfMemory: Insufficient memory is available.
• TimedOut: The operation timed out.

Examples found in repository

examples/pipeline.rs (line 33)

fn pull(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Pull0)?;
    s.listen(url)?;

    loop {
        let msg = s.recv()?;
        let arg = str::from_utf8(&msg).expect("message has invalid UTF-8");

        println!("PULL: RECEIVED \"{}\"", arg);
    }
}

examples/reqrep.rs (line 42)

fn request(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    println!("REQUEST: SENDING DATE REQUEST");
    s.send(DATE_REQUEST.to_le_bytes())?;

    println!("REQUEST: WAITING FOR RESPONSE");
    let msg = s.recv()?;
    let epoch = u64::from_le_bytes(msg[..].try_into().unwrap());

    println!("REQUEST: UNIX EPOCH WAS {} SECONDS AGO", epoch);

    Ok(())
}

/// Run the reply portion of the program.
fn reply(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Rep0)?;
    s.listen(url)?;

    loop {
        println!("REPLY: WAITING FOR COMMAND");
        let mut msg = s.recv()?;

        let cmd = u64::from_le_bytes(msg[..].try_into().unwrap());
        if cmd != DATE_REQUEST {
            println!("REPLY: UNKNOWN COMMAND");
            continue;
        }

        println!("REPLY: RECEIVED DATE REQUEST");
        let rep = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("current system time is before Unix epoch")
            .as_secs();

        msg.clear();
        msg.push_back(&rep.to_le_bytes());

        println!("REPLY: SENDING {}", rep);
        s.send(msg)?;
    }
}

examples/async.rs (line 49)

fn client(url: &str, ms: u64) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    let start = Instant::now();
    s.send(ms.to_le_bytes())?;
    s.recv()?;

    let dur = Instant::now().duration_since(start);
    let subsecs: u64 = dur.subsec_millis().into();
    println!(
        "Request took {} milliseconds",
        dur.as_secs() * 1000 + subsecs
    );

    Ok(())
}

examples/pubsub.rs (line 73)

fn subscriber(url: &str) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Sub0)?;
    s.dial(url)?;

    println!("SUBSCRIBER: SUBSCRIBING TO ALL TOPICS");
    let all_topics = vec![];
    s.set_opt::<Subscribe>(all_topics)?;

    loop {
        let msg = s.recv()?;
        let subs = usize::from_le_bytes(msg[..].try_into().unwrap());
        println!("SUBSCRIBER: THERE ARE {} SUBSCRIBERS", subs);
    }
}

examples/bus.rs (line 41)

fn node(name: &str, listen: &str, dial: &[String]) -> Result<(), Error> {
    let s = Socket::new(Protocol::Bus0)?;
    s.listen(listen)?;

    // Give time for peers to bind.
    thread::sleep(Duration::from_secs(1));
    for peer in dial {
        s.dial(peer)?;
    }

    // SEND
    println!("{0}: SENDING \"{0}\" ONTO BUS", name);
    s.send(name.as_bytes())?;

    // RECV
    loop {
        let msg = s.recv()?;
        let peer = str::from_utf8(&msg).expect("invalid UTF-8");

        println!("{}: RECEIVED \"{}\" FROM BUS", name, peer);
    }
}

examples/survey.rs (line 39)

fn surveyor(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Surveyor0)?;
    s.listen(url)?;

    loop {
        println!("SURVEYOR: SENDING DATE SURVEY REQUEST");
        s.send(DATE.as_bytes())?;

        loop {
            let msg = match s.recv() {
                Ok(m) => m,
                Err(Error::TimedOut) => break,
                Err(e) => return Err(e),
            };

            let date = u64::from_le_bytes(msg[..].try_into().unwrap());
            println!("SURVEYOR: RECEIVED \"{}\" SURVEY RESPONSE", date);
        }

        println!("SURVEYOR SURVEY COMPLETE");
    }
}

/// Respondent socket.
fn respondent(url: &str, name: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Respondent0)?;
    s.dial(url)?;

    loop {
        let mut msg = s.recv()?;

        let survey = str::from_utf8(&msg).expect("invalid UTF-8");
        println!(
            "RESPONDENT ({}): RECEIVED \"{}\" SURVEY REQUEST",
            name, survey
        );

        // Reuse the message to avoid allocation.
        msg.clear();
        let date = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("system time is before Unix epoch")
            .as_secs();

        msg.push_back(&date.to_le_bytes());

        println!("RESPONDENT ({}): SENDING \"{}\"", name, date);
        s.send(msg)?;
    }
}

Additional examples can be found in:

• examples/pair.rs

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

Sends a message on the socket.

The semantics of what sending a message means vary from protocol to protocol, so examination of the protocol documentation is encouraged. For example, with a pub socket the data is broadcast so that any peers who have a suitable subscription will be able to receive it. Furthermore, some protocols may not support sending data (such as sub) or may require other conditions. For example, rep sockets cannot normally send data, which are responses to requests, until they have first received a request.

If the message cannot be sent, then it is returned to the caller as a part of the Error.

Errors

• Closed: The socket is not open.
• IncorrectState: The socket cannot send messages in this state.
• MessageTooLarge: The message is too large.
• NotSupported: The protocol does not support sending messages.
• OutOfMemory: Insufficient memory available.
• TimedOut: The operation timed out.

Examples found in repository

examples/pipeline.rs (line 46)

fn push(url: &str, arg: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Push0)?;
    s.dial(url)?;

    println!("PUSH: SENDING \"{}\"", arg);
    s.send(arg.as_bytes())?;

    // Wait for messages to flush before shutting down.
    thread::sleep(Duration::from_secs(1));
    Ok(())
}

examples/reqrep.rs (line 39)

fn request(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    println!("REQUEST: SENDING DATE REQUEST");
    s.send(DATE_REQUEST.to_le_bytes())?;

    println!("REQUEST: WAITING FOR RESPONSE");
    let msg = s.recv()?;
    let epoch = u64::from_le_bytes(msg[..].try_into().unwrap());

    println!("REQUEST: UNIX EPOCH WAS {} SECONDS AGO", epoch);

    Ok(())
}

/// Run the reply portion of the program.
fn reply(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Rep0)?;
    s.listen(url)?;

    loop {
        println!("REPLY: WAITING FOR COMMAND");
        let mut msg = s.recv()?;

        let cmd = u64::from_le_bytes(msg[..].try_into().unwrap());
        if cmd != DATE_REQUEST {
            println!("REPLY: UNKNOWN COMMAND");
            continue;
        }

        println!("REPLY: RECEIVED DATE REQUEST");
        let rep = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("current system time is before Unix epoch")
            .as_secs();

        msg.clear();
        msg.push_back(&rep.to_le_bytes());

        println!("REPLY: SENDING {}", rep);
        s.send(msg)?;
    }
}

examples/async.rs (line 48)

fn client(url: &str, ms: u64) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Req0)?;
    s.dial(url)?;

    let start = Instant::now();
    s.send(ms.to_le_bytes())?;
    s.recv()?;

    let dur = Instant::now().duration_since(start);
    let subsecs: u64 = dur.subsec_millis().into();
    println!(
        "Request took {} milliseconds",
        dur.as_secs() * 1000 + subsecs
    );

    Ok(())
}

examples/bus.rs (line 37)

fn node(name: &str, listen: &str, dial: &[String]) -> Result<(), Error> {
    let s = Socket::new(Protocol::Bus0)?;
    s.listen(listen)?;

    // Give time for peers to bind.
    thread::sleep(Duration::from_secs(1));
    for peer in dial {
        s.dial(peer)?;
    }

    // SEND
    println!("{0}: SENDING \"{0}\" ONTO BUS", name);
    s.send(name.as_bytes())?;

    // RECV
    loop {
        let msg = s.recv()?;
        let peer = str::from_utf8(&msg).expect("invalid UTF-8");

        println!("{}: RECEIVED \"{}\" FROM BUS", name, peer);
    }
}

examples/survey.rs (line 36)

fn surveyor(url: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Surveyor0)?;
    s.listen(url)?;

    loop {
        println!("SURVEYOR: SENDING DATE SURVEY REQUEST");
        s.send(DATE.as_bytes())?;

        loop {
            let msg = match s.recv() {
                Ok(m) => m,
                Err(Error::TimedOut) => break,
                Err(e) => return Err(e),
            };

            let date = u64::from_le_bytes(msg[..].try_into().unwrap());
            println!("SURVEYOR: RECEIVED \"{}\" SURVEY RESPONSE", date);
        }

        println!("SURVEYOR SURVEY COMPLETE");
    }
}

/// Respondent socket.
fn respondent(url: &str, name: &str) -> Result<(), Error> {
    let s = Socket::new(Protocol::Respondent0)?;
    s.dial(url)?;

    loop {
        let mut msg = s.recv()?;

        let survey = str::from_utf8(&msg).expect("invalid UTF-8");
        println!(
            "RESPONDENT ({}): RECEIVED \"{}\" SURVEY REQUEST",
            name, survey
        );

        // Reuse the message to avoid allocation.
        msg.clear();
        let date = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("system time is before Unix epoch")
            .as_secs();

        msg.push_back(&date.to_le_bytes());

        println!("RESPONDENT ({}): SENDING \"{}\"", name, date);
        s.send(msg)?;
    }
}

examples/pair.rs (line 69)

fn send_recv(s: &Socket, name: &str) -> Result<(), Error> {
    s.set_opt::<RecvTimeout>(Some(Duration::from_millis(100)))?;
    loop {
        // Attempt to reuse the message if we can.
        let mut msg = match s.recv() {
            Ok(m) => {
                let partner = str::from_utf8(&m).expect("invalid UTF-8 message");
                println!("{}: RECEIVED \"{}\"", name, partner);

                m
            }

            Err(Error::TimedOut) => Message::new(),

            Err(e) => return Err(e),
        };

        thread::sleep(Duration::from_secs(1));

        msg.clear();
        write!(msg, "{}", name).expect("failed to write to message");

        println!("{0}: SENDING \"{0}\"", name);
        s.send(msg)?;
    }
}

Additional examples can be found in:

• examples/pubsub.rs

pub fn try_recv(&self) -> Result<Message>

Attempts to receives a message from the socket.

The semantics of what receiving a message means vary from protocol to protocol, so examination of the protocol documentation is encouraged. For example, with a req socket a message may only be received after a request has been sent. Furthermore, some protocols may not support receiving data at all, such as pub.

If no message is available, this function will immediately return.

Errors

• Closed: The socket is not open.
• IncorrectState: The socket cannot receive data in this state.
• NotSupported: The protocol does not support receiving.
• OutOfMemory: Insufficient memory is available.
• TryAgain: The operation would block.

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

Attempts to sends a message on the socket.

The semantics of what sending a message means vary from protocol to protocol, so examination of the protocol documentation is encouraged. For example, with a pub socket the data is broadcast so that any peers who have a suitable subscription will be able to receive it. Furthermore, some protocols may not support sending data (such as sub) or may require other conditions. For example, rep sockets cannot normally send data, which are responses to requests, until they have first received a request.

If the message cannot be sent (e.g., there are no peers or there is backpressure from the peers) then this function will return immediately. If the message cannot be sent, then it is returned to the caller as a part of the Error.

Errors

• Closed: The socket is not open.
• IncorrectState: The socket cannot send messages in this state.
• MessageTooLarge: The message is too large.
• NotSupported: The protocol does not support sending messages.
• OutOfMemory: Insufficient memory available.
• TryAgain: The operation would block.

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

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

Errors

• IncorrectState: The Aio already has a running operation.

pub fn send_async<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.

pub fn pipe_notify<F>(&self, callback: F) -> Result<()>

where F: Fn(Pipe, PipeEvent) + Send + Sync + 'static,

Register a callback function to be called whenever a pipe event occurs on the socket.

Only a single callback function can be supplied at a time. Registering a new callback implicitly unregisters any previously registered.

Errors

None specified.

Panics

If the callback function panics, the program will log the panic if possible and then abort. Future Rustc versions will likely do the same for uncaught panics at FFI boundaries, so this library will produce the abort in order to keep things consistent. As such, the user is responsible for either having a callback that never panics or catching and handling the panic within the callback.

Examples found in repository

examples/pubsub.rs (lines 42-48)

fn publisher(url: &str) -> Result<(), nng::Error> {
    let s = Socket::new(Protocol::Pub0)?;
    let count = Arc::new(AtomicUsize::new(0));
    let count_clone = count.clone();

    s.pipe_notify(move |_, ev| {
        match ev {
            PipeEvent::AddPost => count_clone.fetch_add(1, Ordering::Relaxed),
            PipeEvent::RemovePost => count_clone.fetch_sub(1, Ordering::Relaxed),
            _ => 0,
        };
    })?;

    s.listen(url)?;

    loop {
        // Sleep for a little bit before sending the next message.
        thread::sleep(Duration::from_secs(3));

        // Load the number of subscribers and send the value across
        let data = count.load(Ordering::Relaxed) as u64;
        println!("PUBLISHER: SENDING {}", data);
        s.send(data.to_le_bytes())?;
    }
}

pub fn close(&self)

Close the underlying socket.

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

Closing the socket while data is in transmission will likely lead to loss of that data. There is no automatic linger or flush to ensure that the socket send buffers have completely transmitted. It is recommended to wait a brief period after sending data before calling this function.

This function will be called automatically when all handles have been dropped.

Trait Implementations

impl Clone for Socket

fn clone(&self) -> Socket

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Socket

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Hash for Socket

fn hash<H: Hasher>(&self, state: &mut H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl Ord for Socket

fn cmp(&self, other: &Socket) -> Ordering

This method returns an Ordering between self and other.

fn max(self, other: Self) -> Self

where Self: Sized,

Compares and returns the maximum of two values.

fn min(self, other: Self) -> Self

where Self: Sized,

Compares and returns the minimum of two values.

fn clamp(self, min: Self, max: Self) -> Self

where Self: Sized,

Restrict a value to a certain interval.

impl PartialEq for Socket

fn eq(&self, other: &Socket) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PartialOrd for Socket

fn partial_cmp(&self, other: &Socket) -> Option<Ordering>

This method returns an ordering between self and other values if one exists.

fn lt(&self, other: &Rhs) -> bool

Tests less than (for self and other) and is used by the < operator.

fn le(&self, other: &Rhs) -> bool

Tests less than or equal to (for self and other) and is used by the <= operator.

fn gt(&self, other: &Rhs) -> bool

Tests greater than (for self and other) and is used by the > operator.

fn ge(&self, other: &Rhs) -> bool

Tests greater than or equal to (for self and other) and is used by the >= operator.

impl TryFrom<Socket> for RawSocket

type Error = CookedSocketError

The type returned in the event of a conversion error.

fn try_from(socket: Socket) -> Result<Self, Self::Error>

Performs the conversion.

impl Eq for Socket

impl GetOpt<MaxTtl> for Socket

impl GetOpt<Raw> for Socket

impl GetOpt<RecvBufferSize> for Socket

impl GetOpt<RecvFd> for Socket

impl GetOpt<RecvTimeout> for Socket

impl GetOpt<ResendTime> for Socket

impl GetOpt<SendBufferSize> for Socket

impl GetOpt<SendFd> for Socket

impl GetOpt<SendTimeout> for Socket

impl GetOpt<SocketName> for Socket

impl GetOpt<SurveyTime> for Socket

impl SetOpt<CaFile> for Socket

impl SetOpt<CertKeyFile> for Socket

impl SetOpt<KeepAlive> for Socket

impl SetOpt<MaxTtl> for Socket

impl SetOpt<NoDelay> for Socket

impl SetOpt<ReconnectMaxTime> for Socket

impl SetOpt<ReconnectMinTime> for Socket

impl SetOpt<RecvBufferSize> for Socket

impl SetOpt<RecvMaxSize> for Socket

impl SetOpt<RecvTimeout> for Socket

impl SetOpt<RequestHeaders> for Socket

impl SetOpt<ResendTime> for Socket

impl SetOpt<ResponseHeaders> for Socket

impl SetOpt<SendBufferSize> for Socket

impl SetOpt<SendTimeout> for Socket

impl SetOpt<SocketName> for Socket

impl SetOpt<Subscribe> for Socket

impl SetOpt<SurveyTime> for Socket

impl SetOpt<Unsubscribe> for Socket