Struct nats::jetstream::JetStream

pub struct JetStream { /* private fields */ }

A context for performing JetStream operations.

Implementations

impl JetStream

pub fn new(connection: Connection, options: JetStreamOptions) -> Self

Create a new JetStream context.

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

Publishes a message to JetStream

pub fn publish_with_options(
    &self,
    subject: &str,
    data: impl AsRef<[u8]>,
    options: &PublishOptions
) -> Result<PublishAck>

Publishes a message to JetStream with the given options.

pub fn publish_message(&self, message: &Message) -> Result<PublishAck>

Publishes a Message to JetStream.

pub fn publish_message_with_options(
    &self,
    message: &Message,
    options: &PublishOptions
) -> Result<PublishAck>

Publishes a Message to JetStream with the given options.

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

Create an ephemeral push consumer subscription.

Example

let subscription = context.subscribe("ephemeral")?;
println!("Received message {:?}", subscription.next());

pub fn pull_subscribe(&self, subject: &str) -> Result<PullSubscription>

Creates a pull subscription.

Example

  context.publish("next", b"foo")?;
}
let consumer = context.pull_subscribe("next")?;

consumer.fetch_with_handler(10, |message| {
    println!("received message: {:?}", message);
    Ok(())
})?;

pub fn pull_subscribe_with_options(
    &self,
    subject: &str,
    options: &PullSubscribeOptions
) -> Result<PullSubscription>

Creates a PullSubscription with options.

pub fn subscribe_with_options(
    &self,
    subject: &str,
    options: &SubscribeOptions
) -> Result<PushSubscription>

Creates a push consumer subscription with options.

If said consumer is named and already exists, this will attempt to bind this consumer to that one, else will attempt to create a new internally managed consumer resource.

Example

let sub = js.subscribe_with_options("foo", &SubscribeOptions::bind("existing_stream".to_string(), "existing_consumer".to_string()))?;

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

Creates a push-based consumer subscription with a queue group. The queue group will be used as the durable name.

Example

let subscription = context.queue_subscribe("queue", "queue_group")?;

pub fn queue_subscribe_with_options(
    &self,
    subject: &str,
    queue: &str,
    options: &SubscribeOptions
) -> Result<PushSubscription>

Creates a push-based consumer subscription with a queue group and options.

If a durable name is not set within the options provided options then the queue group will be used as the durable name.

pub fn add_stream<S>(&self, stream_config: S) -> Result<StreamInfo>where
    StreamConfig: From<S>,

Create a JetStream stream.

pub fn update_stream(&self, config: &StreamConfig) -> Result<StreamInfo>

Update a JetStream stream.

pub fn stream_names(&self) -> PagedIterator<'_, String>

List all JetStream stream names. If you also want stream information, use the list_streams method instead.

pub fn list_streams(&self) -> PagedIterator<'_, StreamInfo>

List all JetStream streams.

pub fn list_consumers<S>(
    &self,
    stream: S
) -> Result<PagedIterator<'_, ConsumerInfo>>where
    S: AsRef<str>,

List JetStream consumers for a stream.

pub fn stream_info<S: AsRef<str>>(&self, stream: S) -> Result<StreamInfo>

Query JetStream stream information.

pub fn purge_stream<S: AsRef<str>>(&self, stream: S) -> Result<PurgeResponse>

Purge JetStream stream messages.

pub fn purge_stream_subject<S: AsRef<str>>(
    &self,
    stream: S,
    filter_subject: &str
) -> Result<PurgeResponse>

Purge stream messages matching a subject.

pub fn get_message<S: AsRef<str>>(
    &self,
    stream: S,
    seq: u64
) -> Result<StreamMessage>

Get a message from a stream.

pub fn get_last_message<S: AsRef<str>>(
    &self,
    stream_name: S,
    stream_subject: &str
) -> Result<StreamMessage>

Get the last message from a stream by subject

pub fn delete_message<S: AsRef<str>>(
    &self,
    stream: S,
    sequence_number: u64
) -> Result<bool>

Delete message in a JetStream stream.

pub fn delete_stream<S: AsRef<str>>(&self, stream: S) -> Result<bool>

Delete JetStream stream.

pub fn add_consumer<S, C>(&self, stream: S, config: C) -> Result<ConsumerInfo>where
    S: AsRef<str>,
    ConsumerConfig: From<C>,

Create a JetStream consumer.

pub fn delete_consumer<S, C>(&self, stream: S, consumer: C) -> Result<bool>where
    S: AsRef<str>,
    C: AsRef<str>,

Delete a JetStream consumer.

pub fn consumer_info<S, C>(&self, stream: S, consumer: C) -> Result<ConsumerInfo>where
    S: AsRef<str>,
    C: AsRef<str>,

Query JetStream consumer information.

pub fn account_info(&self) -> Result<AccountInfo>

Query JetStream account information.

impl JetStream

pub fn key_value(&self, bucket: &str) -> Result<Store>

Available on crate feature unstable only.

Bind to an existing key-value store bucket.

Example

context.create_key_value(&Config {
  bucket: "key_value".to_string(),
  ..Default::default()
})?;

let key_value = context.key_value("key_value")?;

pub fn create_key_value(&self, config: &Config) -> Result<Store>

Available on crate feature unstable only.

Create a new key-value store bucket.

Examples

let bucket = context.create_key_value(&Config {
  bucket: "create_key_value".to_string(),
  ..Default::default()
})?;

pub fn delete_key_value(&self, bucket: &str) -> Result<()>

Available on crate feature unstable only.

Delete the specified key value store bucket.

Example

use nats::kv::Config;

context.delete_key_value("delete_key_value")?;

impl JetStream

pub fn create_object_store(&self, config: &Config) -> Result<ObjectStore>

Available on crate feature unstable only.

Creates a new object store bucket.

Example

let bucket = context.create_object_store(&Config {
  bucket: "create_object_store".to_string(),
  ..Default::default()
})?;

pub fn object_store(&self, bucket_name: &str) -> Result<ObjectStore>

Available on crate feature unstable only.

Bind to an existing object store bucket.

Example

context.create_object_store(&Config {
  bucket: "object_store".to_string(),
  ..Default::default()
})?;

let bucket = context.object_store("object_store")?;

pub fn delete_object_store(&self, bucket_name: &str) -> Result<()>

Available on crate feature unstable only.

Delete the underlying stream for the named object.

Example

use nats::object_store::Config;

context.delete_object_store("delete_object_store")?;

Trait Implementations

impl Clone for JetStream

fn clone(&self) -> JetStream

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for JetStream

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

Formats the value using the given formatter.