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
impl Context
pub fn set_timeout(&mut self, timeout: Duration)
sourcepub async fn publish<S: ToSubject>(
&self,
subject: S,
payload: Bytes,
) -> Result<PublishAckFuture, PublishError>
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?;
sourcepub async fn publish_with_headers<S: ToSubject>(
&self,
subject: S,
headers: HeaderMap,
payload: Bytes,
) -> Result<PublishAckFuture, PublishError>
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?;
sourcepub async fn send_publish<S: ToSubject>(
&self,
subject: S,
publish: Publish,
) -> Result<PublishAckFuture, PublishError>
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?;
sourcepub async fn query_account(&self) -> Result<Account, AccountError>
pub async fn query_account(&self) -> Result<Account, AccountError>
Query the server for account information
sourcepub async fn create_stream<S>(
&self,
stream_config: S,
) -> Result<Stream<Info>, CreateStreamError>
pub async fn create_stream<S>( &self, stream_config: S, ) -> Result<Stream<Info>, CreateStreamError>
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?;
sourcepub async fn get_stream_no_info<T: AsRef<str>>(
&self,
stream: T,
) -> Result<Stream<()>, GetStreamError>
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?;
sourcepub async fn get_stream<T: AsRef<str>>(
&self,
stream: T,
) -> Result<Stream, GetStreamError>
pub async fn get_stream<T: AsRef<str>>( &self, stream: T, ) -> Result<Stream, GetStreamError>
sourcepub async fn get_or_create_stream<S>(
&self,
stream_config: S,
) -> Result<Stream, CreateStreamError>
pub async fn get_or_create_stream<S>( &self, stream_config: S, ) -> Result<Stream, CreateStreamError>
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?;
sourcepub async fn delete_stream<T: AsRef<str>>(
&self,
stream: T,
) -> Result<DeleteStatus, DeleteStreamError>
pub async fn delete_stream<T: AsRef<str>>( &self, stream: T, ) -> Result<DeleteStatus, DeleteStreamError>
sourcepub async fn update_stream<S>(
&self,
config: S,
) -> Result<Info, UpdateStreamError>
pub async fn update_stream<S>( &self, config: S, ) -> Result<Info, UpdateStreamError>
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?;
sourcepub async fn stream_by_subject<T: Into<String>>(
&self,
subject: T,
) -> Result<String, GetStreamByNameError>
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.>");
sourcepub fn stream_names(&self) -> StreamNames
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);
}
sourcepub fn streams(&self) -> Streams
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);
}
sourcepub async fn get_key_value<T: Into<String>>(
&self,
bucket: T,
) -> Result<Store, KeyValueError>
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?;
sourcepub async fn create_key_value(
&self,
config: Config,
) -> Result<Store, CreateKeyValueError>
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?;
sourcepub async fn delete_key_value<T: AsRef<str>>(
&self,
bucket: T,
) -> Result<DeleteStatus, KeyValueError>
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?;
sourcepub async fn get_consumer_from_stream<T, C, S>(
&self,
consumer: C,
stream: S,
) -> Result<Consumer<T>, ConsumerError>
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?;
sourcepub async fn delete_consumer_from_stream<C: AsRef<str>, S: AsRef<str>>(
&self,
consumer: C,
stream: S,
) -> Result<DeleteStatus, ConsumerError>
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?;
sourcepub async fn create_consumer_on_stream<C: IntoConsumerConfig + FromConsumer, S: AsRef<str>>(
&self,
config: C,
stream: S,
) -> Result<Consumer<C>, ConsumerError>
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?;
sourcepub async fn update_consumer_on_stream<C: IntoConsumerConfig + FromConsumer, S: AsRef<str>>(
&self,
config: C,
stream: S,
) -> Result<Consumer<C>, ConsumerUpdateError>
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?;
sourcepub async fn create_consumer_strict_on_stream<C: IntoConsumerConfig + FromConsumer, S: AsRef<str>>(
&self,
config: C,
stream: S,
) -> Result<Consumer<C>, ConsumerCreateStrictError>
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?;
sourcepub async fn request<S, T, V>(
&self,
subject: S,
payload: &T,
) -> Result<V, RequestError>
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?;
sourcepub async fn create_object_store(
&self,
config: Config,
) -> Result<ObjectStore, CreateObjectStoreError>
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?;
sourcepub async fn get_object_store<T: AsRef<str>>(
&self,
bucket_name: T,
) -> Result<ObjectStore, ObjectStoreError>
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?;
sourcepub async fn delete_object_store<T: AsRef<str>>(
&self,
bucket_name: T,
) -> Result<(), DeleteObjectStore>
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§
Auto Trait Implementations§
impl Freeze for Context
impl !RefUnwindSafe for Context
impl Send for Context
impl Sync for Context
impl Unpin for Context
impl !UnwindSafe for Context
Blanket Implementations§
source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
source§impl<T> CloneToUninit for Twhere
T: Clone,
impl<T> CloneToUninit for Twhere
T: Clone,
source§unsafe fn clone_to_uninit(&self, dst: *mut T)
unsafe fn clone_to_uninit(&self, dst: *mut T)
clone_to_uninit
)