Struct Context

pub struct Context { /* private fields */ }

A context which can perform jetstream scoped requests.

Implementations

impl Context

pub fn set_timeout(&mut self, timeout: Duration)

pub async fn publish<S: ToSubject>( &self, subject: S, payload: Bytes, ) -> Result<PublishAckFuture, PublishError>

Publishes jetstream::Message to the Stream without waiting for acknowledgment from the server that the message has been successfully delivered.

Acknowledgment future that can be polled is returned instead.

If the stream does not exist, no responders error will be returned.

Examples

Publish, and after each publish, await for acknowledgment.

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let ack = jetstream.publish("events", "data".into()).await?;
ack.await?;
jetstream.publish("events", "data".into()).await?.await?;

Publish and do not wait for the acknowledgment. Await can be deferred to when needed or ignored entirely.

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let first_ack = jetstream.publish("events", "data".into()).await?;
let second_ack = jetstream.publish("events", "data".into()).await?;
first_ack.await?;
second_ack.await?;

pub async fn publish_with_headers<S: ToSubject>( &self, subject: S, headers: HeaderMap, payload: Bytes, ) -> Result<PublishAckFuture, PublishError>

Publish a message with headers to a given subject associated with a stream and returns an acknowledgment from the server that the message has been successfully delivered.

If the stream does not exist, no responders error will be returned.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let mut headers = async_nats::HeaderMap::new();
headers.append("X-key", "Value");
let ack = jetstream
    .publish_with_headers("events", headers, "data".into())
    .await?;

pub async fn send_publish<S: ToSubject>( &self, subject: S, publish: Publish, ) -> Result<PublishAckFuture, PublishError>

Publish a message built by Publish and returns an acknowledgment future.

If the stream does not exist, no responders error will be returned.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let ack = jetstream
    .send_publish(
        "events",
        Publish::build().payload("data".into()).message_id("uuid"),
    )
    .await?;

pub async fn query_account(&self) -> Result<Account, AccountError>

Query the server for account information

pub async fn create_stream<S>( &self, stream_config: S, ) -> Result<Stream<Info>, CreateStreamError>

where Config: From<S>,

Create a JetStream Stream with given config and return a handle to it. That handle can be used to manage and use Consumer.

Examples

use async_nats::jetstream::stream::Config;
use async_nats::jetstream::stream::DiscardPolicy;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream
    .create_stream(Config {
        name: "events".to_string(),
        max_messages: 100_000,
        discard: DiscardPolicy::Old,
        ..Default::default()
    })
    .await?;

pub async fn get_stream_no_info<T: AsRef<str>>( &self, stream: T, ) -> Result<Stream<()>, GetStreamError>

Checks for Stream existence on the server and returns handle to it. That handle can be used to manage and use Consumer. This variant does not fetch Stream info from the server. It means it does not check if the stream actually exists. If you run more operations on few streams, it is better to use Context::get_stream instead. If you however run single operations on many streams, this method is more efficient.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.get_stream_no_info("events").await?;

pub async fn get_stream<T: AsRef<str>>( &self, stream: T, ) -> Result<Stream, GetStreamError>

Checks for Stream existence on the server and returns handle to it. That handle can be used to manage and use Consumer.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.get_stream("events").await?;

pub async fn get_or_create_stream<S>( &self, stream_config: S, ) -> Result<Stream, CreateStreamError>

where S: Into<Config>,

Create a stream with the given configuration on the server if it is not present. Returns a handle to the stream on the server.

Note: This does not validate if the Stream on the server is compatible with the configuration passed in.

Examples

use async_nats::jetstream::stream::Config;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream
    .get_or_create_stream(Config {
        name: "events".to_string(),
        max_messages: 10_000,
        ..Default::default()
    })
    .await?;

pub async fn delete_stream<T: AsRef<str>>( &self, stream: T, ) -> Result<DeleteStatus, DeleteStreamError>

Deletes a Stream with a given name.

Examples

use async_nats::jetstream::stream::Config;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream.delete_stream("events").await?;

pub async fn update_stream<S>( &self, config: S, ) -> Result<Info, UpdateStreamError>

where S: Borrow<Config>,

Updates a Stream with a given config. If specific field cannot be updated, error is returned.

Examples

use async_nats::jetstream::stream::Config;
use async_nats::jetstream::stream::DiscardPolicy;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let stream = jetstream
    .update_stream(&Config {
        name: "events".to_string(),
        discard: DiscardPolicy::New,
        max_messages: 50_000,
        ..Default::default()
    })
    .await?;

pub async fn stream_by_subject<T: Into<String>>( &self, subject: T, ) -> Result<String, GetStreamByNameError>

Looks up Stream that contains provided subject.

Examples

use futures::TryStreamExt;
let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let stream_name = jetstream.stream_by_subject("foo.>");

pub fn stream_names(&self) -> StreamNames

Lists names of all streams for current context.

Examples

use futures::TryStreamExt;
let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let mut names = jetstream.stream_names();
while let Some(stream) = names.try_next().await? {
    println!("stream: {}", stream);
}

pub fn streams(&self) -> Streams

Lists all streams info for current context.

Examples

use futures::TryStreamExt;
let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let mut streams = jetstream.streams();
while let Some(stream) = streams.try_next().await? {
    println!("stream: {:?}", stream);
}

pub async fn get_key_value<T: Into<String>>( &self, bucket: T, ) -> Result<Store, KeyValueError>

Returns an existing key-value bucket.

Examples

let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let kv = jetstream.get_key_value("bucket").await?;

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

Creates a new key-value bucket.

Examples

let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let kv = jetstream
    .create_key_value(async_nats::jetstream::kv::Config {
        bucket: "kv".to_string(),
        history: 10,
        ..Default::default()
    })
    .await?;

pub async fn delete_key_value<T: AsRef<str>>( &self, bucket: T, ) -> Result<DeleteStatus, KeyValueError>

Deletes given key-value bucket.

Examples

let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let kv = jetstream
    .create_key_value(async_nats::jetstream::kv::Config {
        bucket: "kv".to_string(),
        history: 10,
        ..Default::default()
    })
    .await?;

pub async fn get_consumer_from_stream<T, C, S>( &self, consumer: C, stream: S, ) -> Result<Consumer<T>, ConsumerError>

where T: FromConsumer + IntoConsumerConfig, S: AsRef<str>, C: AsRef<str>,

Get a crate::jetstream::consumer::Consumer straight from Context, without binding to a Stream first.

It has one less interaction with the server when binding to only one crate::jetstream::consumer::Consumer.

Examples:

use async_nats::jetstream::consumer::PullConsumer;

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_consumer_from_stream("consumer", "stream")
    .await?;

pub async fn delete_consumer_from_stream<C: AsRef<str>, S: AsRef<str>>( &self, consumer: C, stream: S, ) -> Result<DeleteStatus, ConsumerError>

Delete a crate::jetstream::consumer::Consumer straight from Context, without binding to a Stream first.

It has one less interaction with the server when binding to only one crate::jetstream::consumer::Consumer.

Examples:

use async_nats::jetstream::consumer::PullConsumer;

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

jetstream
    .delete_consumer_from_stream("consumer", "stream")
    .await?;

pub async fn create_consumer_on_stream<C: IntoConsumerConfig + FromConsumer, S: AsRef<str>>( &self, config: C, stream: S, ) -> Result<Consumer<C>, ConsumerError>

Create or update a Durable or Ephemeral Consumer (if durable_name was not provided) and returns the info from the server about created Consumer without binding to a Stream first. If you want a strict update or create, use Context::create_consumer_strict_on_stream or Context::update_consumer_on_stream.

Examples

use async_nats::jetstream::consumer;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: consumer::PullConsumer = jetstream
    .create_consumer_on_stream(
        consumer::pull::Config {
            durable_name: Some("pull".to_string()),
            ..Default::default()
        },
        "stream",
    )
    .await?;

pub async fn update_consumer_on_stream<C: IntoConsumerConfig + FromConsumer, S: AsRef<str>>( &self, config: C, stream: S, ) -> Result<Consumer<C>, ConsumerUpdateError>

Available on crate feature server_2_10 only.

Update an existing consumer. This call will fail if the consumer does not exist. returns the info from the server about updated Consumer without binding to a Stream first.

Examples

use async_nats::jetstream::consumer;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: consumer::PullConsumer = jetstream
    .update_consumer_on_stream(
        consumer::pull::Config {
            durable_name: Some("pull".to_string()),
            description: Some("updated pull consumer".to_string()),
            ..Default::default()
        },
        "stream",
    )
    .await?;

pub async fn create_consumer_strict_on_stream<C: IntoConsumerConfig + FromConsumer, S: AsRef<str>>( &self, config: C, stream: S, ) -> Result<Consumer<C>, ConsumerCreateStrictError>

Available on crate feature server_2_10 only.

Create consumer on stream, but only if it does not exist or the existing config is exactly the same. This method will fail if consumer is already present with different config. returns the info from the server about created Consumer without binding to a Stream first.

Examples

use async_nats::jetstream::consumer;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: consumer::PullConsumer = jetstream
    .create_consumer_strict_on_stream(
        consumer::pull::Config {
            durable_name: Some("pull".to_string()),
            ..Default::default()
        },
        "stream",
    )
    .await?;

pub async fn request<S, T, V>( &self, subject: S, payload: &T, ) -> Result<V, RequestError>

where S: ToSubject, T: ?Sized + Serialize, V: DeserializeOwned,

Send a request to the jetstream JSON API.

This is a low level API used mostly internally, that should be used only in specific cases when this crate API on Consumer or Stream does not provide needed functionality.

Examples

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let response: Response<Info> = jetstream.request("STREAM.INFO.events", &()).await?;

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

Creates a new object store bucket.

Examples

let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);
let bucket = jetstream
    .create_object_store(async_nats::jetstream::object_store::Config {
        bucket: "bucket".to_string(),
        ..Default::default()
    })
    .await?;

pub async fn get_object_store<T: AsRef<str>>( &self, bucket_name: T, ) -> Result<ObjectStore, ObjectStoreError>

Get an existing object store bucket.

Examples

let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);
let bucket = jetstream.get_object_store("bucket").await?;

pub async fn delete_object_store<T: AsRef<str>>( &self, bucket_name: T, ) -> Result<(), DeleteObjectStore>

Delete a object store bucket.

Examples

let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);
let bucket = jetstream.delete_object_store("bucket").await?;

Trait Implementations

impl Clone for Context

fn clone(&self) -> Context

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for Context

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

Formats the value using the given formatter.