Struct async_nats::jetstream::context::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(
    &self,
    subject: String,
    payload: Bytes
) -> Result<PublishAckFuture, Error>

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".to_string(), "data".into()).await?;
ack.await?;
jetstream.publish("events".to_string(), "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".to_string(), "data".into()).await?;
let second_ack = jetstream.publish("events".to_string(), "data".into()).await?;
first_ack.await?;
second_ack.await?;

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

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".to_string(), headers, "data".into()).await?;

pub async fn send_publish(
    &self,
    subject: String,
    publish: Publish
) -> Result<PublishAckFuture, Error>

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".to_string(),
    Publish::build().payload("data".into()).message_id("uuid")
).await?;

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

Query the server for account information

pub async fn create_stream<S>(&self, stream_config: S) -> Result<Stream, Error>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<T: AsRef<str>>(&self, stream: T) -> Result<Stream, Error>

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, Error>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, Error>

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, Error>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 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, Error>

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, Error>

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, Error>

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 request<T, V>(
    &self,
    subject: String,
    payload: &T
) -> Result<Response<V>, Error>where
    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".to_string(), &()).await?;

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

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, Error>

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.get_object_store("bucket").await?;

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

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.