Struct nats::Subscription

source ·

pub struct Subscription(/* private fields */);

Expand description

A Subscription receives Messages published to specific NATS Subjects.

Implementations§

source §

impl Subscription

source

pub fn receiver(&self) -> &Receiver<Message>

Get a crossbeam Receiver for subscription messages. Useful for crossbeam_channel::select macro

§Example

let sub1_ch = sub1.receiver();
let sub2_ch = sub2.receiver();
crossbeam_channel::select! {
    recv(sub1_ch) -> msg => {
        println!("Got message from sub1: {:?}", msg);
        Ok(())
    }
    recv(sub2_ch) -> msg => {
        println!("Got message from sub2: {:?}", msg);
        Ok(())
    }
}

source

pub fn next(&self) -> Option<Message>

Get the next message, or None if the subscription has been unsubscribed or the connection closed.

§Example

if let Some(msg) = sub.next() {}

source

pub fn try_next(&self) -> Option<Message>

Try to get the next message, or None if no messages are present or if the subscription has been unsubscribed or the connection closed.

§Example

if let Some(msg) = sub.try_next() {
    println!("Received {}", msg);
}

source

pub fn next_timeout(&self, timeout: Duration) -> Result<Message>

Get the next message, or a timeout error if no messages are available for timeout.

§Example

if let Ok(msg) = sub.next_timeout(std::time::Duration::from_secs(1)) {}

source

pub fn messages(&self) -> Iter<'_>

Returns a blocking message iterator. Same as calling iter().

§Example

for msg in sub.messages() {}

source

pub fn iter(&self) -> Iter<'_>

Returns a blocking message iterator.

§Example

for msg in sub.iter() {}

source

pub fn try_iter(&self) -> TryIter<'_>

Returns a non-blocking message iterator.

§Example

for msg in sub.try_iter() {}

source

pub fn timeout_iter(&self, timeout: Duration) -> TimeoutIter<'_>

Returns a blocking message iterator with a time deadline for blocking.

§Example

for msg in sub.timeout_iter(std::time::Duration::from_secs(1)) {}

source

pub fn with_handler<F>(self, handler: F) -> Handler
where F: Fn(Message) -> Result<()> + Send + 'static,

Attach a closure to handle messages. This closure will execute in a separate thread. The result of this call is a Handler which can not be iterated and must be unsubscribed or closed directly to unregister interest. A Handler will not unregister interest with the server when drop(&mut self) is called.

§Example

nc.subscribe("bar")?.with_handler(move |msg| {
    println!("Received {}", &msg);
    Ok(())
});

source

pub fn set_message_limits(&self, limit: usize)

Sets limit of how many messages can wait in internal queue. If limit will be reached, error_callback will be fired with information which subscription is affected

§Example

let sub = nc.subscribe("bar")?;
sub.set_message_limits(1000);

source

pub fn dropped_messages(&self) -> Result<usize>

Returns number of dropped messages for this Subscription. Dropped messages occur when set_message_limits is set and threshold is reached, triggering slow consumer error.

§Example:

let sub = nc.subscribe("bar")?;
sub.set_message_limits(1000);
println!("dropped messages: {}", sub.dropped_messages()?);

source

pub fn unsubscribe(self) -> Result<()>

Unsubscribe a subscription immediately without draining. Use drain instead if you want any pending messages to be processed by a handler, if one is configured.

§Example

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

source

pub fn close(self) -> Result<()>

Close a subscription. Same as unsubscribe

Use drain instead if you want any pending messages to be processed by a handler, if one is configured.

§Example

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

source

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

Send an unsubscription 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 on the Connection struct which will drain all subscriptions for the NATS client, and transition the entire system into the closed state afterward.

§Example


let mut sub = nc.subscribe("test.drain")?;

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

let mut received = false;
for _ in sub {
    received = true;
}

assert!(received);