Struct wasmcloud_control_interface::kv::CachedKvStore

pub struct CachedKvStore { /* private fields */ }

Implementations

impl CachedKvStore

pub async fn new( nc: Client, lattice_prefix: &str, js_domain: Option<String> ) -> Result<Self, Box<dyn Error + Send + Sync>>

Create a new KV store with the given configuration. This function will do an initial fetch of all claims and linkdefs from the store and then start a watcher to keep the cache up to date. All data fetched from this store will be from the in memory cache

Methods from Deref<Target = Store>

pub async fn status(&self) -> impl Future<Output = Result<Status, StatusError>>

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>( &self, key: T, value: Bytes ) -> impl Future<Output = Result<u64, PutError>>where T: AsRef<str>,

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>( &self, key: T ) -> impl Future<Output = Result<Option<Entry>, EntryError>>where T: Into<String>,

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>( &self, key: T ) -> impl Future<Output = Result<Watch<'_>, WatchError>>where T: AsRef<str>,

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_with_history<T>( &self, key: T ) -> impl Future<Output = Result<Watch<'_>, WatchError>>where T: AsRef<str>,

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

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

pub async fn watch_all( &self ) -> impl Future<Output = Result<Watch<'_>, WatchError>>

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>( &self, key: T ) -> impl Future<Output = Result<Option<Bytes>, EntryError>>where T: Into<String>,

Retrieves the 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 value = kv.get("key").await?;
match value {
    Some(bytes) => {
        let value_str = std::str::from_utf8(&bytes)?;
        println!("Value: {}", value_str);
    }
    None => {
        println!("Key not found or value not set");
    }
}

pub async fn update<T>( &self, key: T, value: Bytes, revision: u64 ) -> impl Future<Output = Result<u64, UpdateError>>where T: AsRef<str>,

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>( &self, key: T ) -> impl Future<Output = Result<(), UpdateError>>where T: AsRef<str>,

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>( &self, key: T ) -> impl Future<Output = Result<(), UpdateError>>where T: AsRef<str>,

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>( &self, key: T ) -> impl Future<Output = Result<History<'_>, WatchError>>where T: AsRef<str>,

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) -> impl Future<Output = Result<Keys<'_>, WatchError>>

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 AsRef<Store> for CachedKvStore

fn as_ref(&self) -> &Store

Converts this type into a shared reference of the (usually inferred) input type.

impl Clone for CachedKvStore

fn clone(&self) -> CachedKvStore

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for CachedKvStore

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

Formats the value using the given formatter.

impl Deref for CachedKvStore

type Target = Store

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl Drop for CachedKvStore

fn drop(&mut self)

Executes the destructor for this type.

impl KvStore for CachedKvStore

fn get_links<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<Vec<LinkDefinition>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Return a copy of all link definitions in the store

fn get_all_claims<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, String>>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Return a copy of all claims in the store

fn get_provider_claims<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, String>>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Return a copy of all provider claims in the store

fn get_actor_claims<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, String>>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Return a copy of all actor claims in the store

fn get_filtered_links<'life0, 'async_trait, F>( &'life0 self, filter_fn: F ) -> Pin<Box<dyn Future<Output = Result<Vec<LinkDefinition>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where F: FnMut(&LinkDefinition) -> bool + Send + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

A convenience function to get a list of link definitions filtered using the given filter function

fn get_link<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, actor_id: &'life1 str, link_name: &'life2 str, contract_id: &'life3 str ) -> Pin<Box<dyn Future<Output = Result<Option<LinkDefinition>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Get a link definition for a specific ID (actor_id, contract_id, link_name)

fn get_claims<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str ) -> Pin<Box<dyn Future<Output = Result<Option<HashMap<String, String>>, Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get claims for a specific provider or actor id

fn put_link<'life0, 'async_trait>( &'life0 self, ld: LinkDefinition ) -> Pin<Box<dyn Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait,

Adds a link definition to the store

fn delete_link<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, actor_id: &'life1 str, contract_id: &'life2 str, link_name: &'life3 str ) -> Pin<Box<dyn Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'async_trait>>where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Deletes a link definition from the store