Struct nats::jetstream::push_subscription::PushSubscription

source ·

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

Expand description

A PushSubscription receives Messages published to specific NATS Subjects.

Implementations§

source §

impl PushSubscription

source

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

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

§Example


if let Some(message) = subscription.next() {
    println!("Received {}", message);
}

source

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

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

§Example

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

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(message) = subscription.next_timeout(std::time::Duration::from_secs(1)) {
    println!("Received {}", message);
}

source

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

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

§Example

for message in subscription.messages() {
    println!("Received message {:?}", message);
}

source

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

Returns a blocking message iterator.

§Example

for message in subscription.iter() {}

source

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

Returns a non-blocking message iterator.

§Example

for message in subscription.try_iter() {}

source

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

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

§Example

for message in subscription.timeout_iter(std::time::Duration::from_secs(1)) {
    println!("Received message {:?}", message);
}

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

context
    .subscribe("with_handler")?
    .with_handler(move |message| {
        println!("received {}", &message);
        Ok(())
    });

source

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

Attach a closure to process and acknowledge 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

context
    .subscribe("with_process_handler")?
    .with_process_handler(|message| {
        println!("Received {}", &message);

        Ok(())
    });

source

pub fn process<R, F: Fn(&Message) -> Result<R>>(&mut self, f: F) -> Result<R>

Process and acknowledge a single message, waiting indefinitely for one to arrive.

Does not acknowledge the processed message if the closure returns an Err.

Example

let mut subscription = context.subscribe("process")?;
subscription.process(|message| {
    println!("Received message {:?}", message);

    Ok(())
})?;

source

pub fn process_timeout<R, F: Fn(&Message) -> Result<R>>( &mut self, timeout: Duration, f: F ) -> Result<R>

Process and acknowledge a single message, waiting up to timeout configured timeout before returning a timeout error.

Does not ack the processed message if the internal closure returns an Err.

Example

let mut subscription = context.subscribe("process_timeout")?;
subscription.process_timeout(std::time::Duration::from_secs(1), |message| {
    println!("Received message {:?}", message);

    Ok(())
})?;

source

pub fn consumer_info(&self) -> Result<ConsumerInfo>

Sends a request to fetch current information about the target consumer.

§Example

let subscription = context.subscribe("consumer_info")?;
let info = subscription.consumer_info()?;

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 subscription = context.subscribe("unsubscribe")?;
subscription.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 subscription = context.subscribe("close")?;
subscription.close()?;

source

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

Send an unsubscription and flush the connection, allowing any unprocessed messages to be handled by a Subscription

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 subscription = context.subscribe("drain")?;

context.publish("drain", "foo")?;
context.publish("drain", "bar")?;
context.publish("drain", "baz")?;

subscription.drain()?;

assert!(subscription.next().is_some());