Struct nats::Connection

pub struct Connection(_);

A NATS connection.

Implementations

impl Connection

pub fn subscribe(&self, subject: &str) -> Result<Subscription>

Create a subscription for the given NATS connection.

Example

let sub = nc.subscribe("foo")?;

pub fn queue_subscribe(
    &self,
    subject: &str,
    queue: &str
) -> Result<Subscription>

Create a queue subscription for the given NATS connection.

Example

let sub = nc.queue_subscribe("foo", "production")?;

pub fn publish(&self, subject: &str, msg: impl AsRef<[u8]>) -> Result<()>

Publish a message on the given subject.

Example

nc.publish("foo", "Hello World!")?;

pub fn publish_request(
    &self,
    subject: &str,
    reply: &str,
    msg: impl AsRef<[u8]>
) -> Result<()>

Publish a message on the given subject with a reply subject for responses.

Example

let reply = nc.new_inbox();
let rsub = nc.subscribe(&reply)?;
nc.publish_request("foo", &reply, "Help me!")?;

pub fn new_inbox(&self) -> String

Create a new globally unique inbox which can be used for replies.

Example

let reply = nc.new_inbox();
let rsub = nc.subscribe(&reply)?;

pub fn request(&self, subject: &str, msg: impl AsRef<[u8]>) -> Result<Message>

Publish a message on the given subject as a request and receive the response.

Example

let resp = nc.request("foo", "Help me?")?;

pub fn request_timeout(
    &self,
    subject: &str,
    msg: impl AsRef<[u8]>,
    timeout: Duration
) -> Result<Message>

Publish a message on the given subject as a request and receive the response. This call will return after the timeout duration if no response is received.

Example

let resp = nc.request_timeout("foo", "Help me?", std::time::Duration::from_secs(2))?;

pub fn request_multi(
    &self,
    subject: &str,
    msg: impl AsRef<[u8]>
) -> Result<Subscription>

Publish a message on the given subject as a request and allow multiple responses.

Example

for msg in nc.request_multi("foo", "Help")?.iter().take(1) {}

pub fn flush(&self) -> Result<()>

Flush a NATS connection by sending a PING protocol and waiting for the responding PONG. Will fail with TimedOut if the server does not respond with in 10 seconds. Will fail with NotConnected if the server is not currently connected. Will fail with BrokenPipe if the connection to the server is lost.

Example

nc.flush()?;

pub fn flush_timeout(&self, duration: Duration) -> Result<()>

Flush a NATS connection by sending a PING protocol and waiting for the responding PONG. Will fail with TimedOut if the server takes longer than this duration to respond. Will fail with NotConnected if the server is not currently connected. Will fail with BrokenPipe if the connection to the server is lost.

Example

nc.flush()?;

pub fn close(self)

Close a NATS connection. All clones of this Connection will also be closed, as the backing IO threads are shared.

If the client is currently connected to a server, the outbound write buffer will be flushed in the process of shutting down.

Example

nc.close();

pub fn rtt(&self) -> Result<Duration>

Calculates the round trip time between this client and the server, if the server is currently connected. Fails with TimedOut if the server takes more than 10 seconds to respond.

Example

println!("server rtt: {:?}", nc.rtt());

pub fn client_ip(&self) -> Result<IpAddr>

Returns the client IP as known by the server. Supported as of server version 2.1.6.

Example

println!("ip: {:?}", nc.client_ip());

pub fn client_id(&self) -> u64

Returns the client ID as known by the most recently connected server.

Example

println!("ip: {:?}", nc.client_id());

pub fn drain(&self) -> Result<()>

Send an unsubscription for all subs then flush the connection, allowing any unprocessed messages to be handled by a handler function if one is configured.

After the flush returns, we know that a round-trip to the server has happened after it received our unsubscription, so we shut down the subscriber afterwards.

A similar method exists for the Subscription struct which will drain a single Subscription without shutting down the entire connection afterward.

Example

let received = Arc::new(AtomicBool::new(false));
let received_2 = received.clone();

nc.subscribe("test.drain")?.with_handler(move |m| {
    received_2.store(true, SeqCst);
    Ok(())
});

nc.publish("test.drain", "message")?;
nc.drain()?;


assert!(received.load(SeqCst));

pub fn publish_with_reply_or_headers(
    &self,
    subject: &str,
    reply: Option<&str>,
    headers: Option<&Headers>,
    msg: impl AsRef<[u8]>
) -> Result<()>

Publish a message which may have a reply subject or headers set.

Example

let sub = nc.subscribe("foo.headers")?;
let headers = [("header1", "value1"),
               ("header2", "value2")].iter().collect();
let reply_to = None;
nc.publish_with_reply_or_headers("foo.headers", reply_to, Some(&headers), "Hello World!")?;
nc.flush()?;
let message = sub.next_timeout(std::time::Duration::from_secs(2)).unwrap();
assert_eq!(message.headers.unwrap().len(), 2);

Trait Implementations

impl Clone for Connection

pub fn clone(&self) -> Connection

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for Connection

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

Formats the value using the given formatter.