Struct async_nats::jetstream::context::Context

source ·
pub struct Context { /* private fields */ }
Expand description

A context which can perform jetstream scoped requests.

Implementations§

source§

impl Context

source

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

source

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?;
source

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?;
source

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?;
source

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

Query the server for account information

source

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?;
source

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("events").await?;
source

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?;
source

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?;
source

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?;
source

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?;
source

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.>");
source

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);
}
source

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);
}
source

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?;
source

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?;
source

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?;
source

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

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?;
source

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?;
source

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?;
source

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

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?;
source

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

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?;
source

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

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?;
source

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?;
source

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?;
source

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§

source§

impl Clone for Context

source§

fn clone(&self) -> Context

Returns a copy of the value. Read more
1.0.0 · source§

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

Performs copy-assignment from source. Read more
source§

impl Debug for Context

source§

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> CloneToUninit for T
where T: Clone,

source§

unsafe fn clone_to_uninit(&self, dst: *mut T)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dst. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T> Instrument for T

source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<T> Same for T

source§

type Output = T

Should always be Self
source§

impl<T> ToOwned for T
where T: Clone,

source§

type Owned = T

The resulting type after obtaining ownership.
source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

source§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

source§

fn vzip(self) -> V

source§

impl<T> WithSubscriber for T

source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more