Struct async_nats::jetstream::kv::Store

pub struct Store {
    pub name: String,
    pub stream_name: String,
    pub prefix: String,
    pub put_prefix: Option<String>,
    pub use_jetstream_prefix: bool,
    pub stream: Stream,
}

A struct used as a handle for the bucket.

Fields

name: String

stream_name: String

prefix: String

put_prefix: Option<String>

use_jetstream_prefix: bool

stream: Stream

Implementations

impl Store

pub async fn status(&self) -> Result<Status, Error>

Queries the server and returns status from the server.

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?;
let status = kv.status().await?;
println!("status: {:?}", status);

pub async fn put<T: AsRef<str>>( &self, key: T, value: Bytes ) -> Result<u64, Error>

Puts new key value pair into the bucket. If key didn’t exist, it is created. If it did exist, a new value with a new version is added.

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?;
let status = kv.put("key", "value".into()).await?;

pub async fn entry<T: Into<String>>( &self, key: T ) -> Result<Option<Entry>, Error>

Retrieves the last Entry for a given key from a 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?;
let status = kv.put("key", "value".into()).await?;
let entry = kv.entry("key").await?;
println!("entry: {:?}", entry);

pub async fn watch<T: AsRef<str>>(&self, key: T) -> Result<Watch<'_>, Error>

Creates a [futures::Stream] over Entries a given key in the bucket, which yields values whenever there are changes for that key.

Examples

use futures::StreamExt;
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?;
let mut entries = kv.watch("kv").await?;
while let Some(entry) = entries.next().await {
    println!("entry: {:?}", entry);
}

pub async fn watch_all(&self) -> Result<Watch<'_>, Error>

Creates a [futures::Stream] over Entries for all keys, which yields values whenever there are changes in the bucket.

Examples

use futures::StreamExt;
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?;
let mut entries = kv.watch_all().await?;
while let Some(entry) = entries.next().await {
    println!("entry: {:?}", entry);
}

pub async fn get<T: Into<String>>( &self, key: T ) -> Result<Option<Vec<u8>>, Error>

pub async fn update<T: AsRef<str>>( &self, key: T, value: Bytes, revision: u64 ) -> Result<u64, Error>

Updates a value for a given key, but only if passed revision is the last revision in the bucket.

Examples

use futures::StreamExt;
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?;
let revision = kv.put("key", "value".into()).await?;
kv.update("key", "updated".into(), revision).await?;

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

Deletes a given key. This is a non-destructive operation, which sets a DELETE marker.

Examples

use futures::StreamExt;
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?;
kv.put("key", "value".into()).await?;
kv.delete("key").await?;

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

Purges all the revisions of a entry destructively, leaving behind a single purge entry in-place.

Examples

use futures::StreamExt;
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?;
kv.put("key", "value".into()).await?;
kv.put("key", "another".into()).await?;
kv.purge("key").await?;

pub async fn history<T: AsRef<str>>(&self, key: T) -> Result<History<'_>, Error>

Returns a [futures::Stream] that allows iterating over all Operations that happen for given key.

Examples

use futures::StreamExt;
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?;
let mut entries = kv.history("kv").await?;
while let Some(entry) = entries.next().await {
    println!("entry: {:?}", entry);
}

pub async fn keys(&self) -> Result<Keys<'_>, Error>

Returns a [futures::Stream] that allows iterating over all keys in the bucket.

Examples

Iterating over each each key individually

use futures::{StreamExt, TryStreamExt};
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?;
let mut keys = kv.keys().await?.boxed();
while let Some(key) = keys.try_next().await? {
    println!("key: {:?}", key);
}

Collecting it into a vector of keys

use futures::TryStreamExt;
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?;
let keys = kv.keys().await?.try_collect::<Vec<String>>().await?;
println!("Keys: {:?}", keys);

Trait Implementations

impl Clone for Store

fn clone(&self) -> Store

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for Store

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

Formats the value using the given formatter.