Trait StateStore 

pub trait StateStore: Any + Send {
    // Required methods
    fn name(&self) -> &str;
    fn flush<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn close(&mut self);
    fn as_any_mut(&mut self) -> &mut dyn Any;
    fn changelog_topic(&self) -> &str;
    fn take_changelog(&mut self) -> Vec<(Bytes, Option<Bytes>)>;
    fn apply_changelog<'life0, 'async_trait>(
        &'life0 mut self,
        key: Bytes,
        value: Option<Bytes>,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn set_logging(&mut self, on: bool);
    fn clear<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn snapshot<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Bytes> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn restore_snapshot<'life0, 'async_trait>(
        &'life0 mut self,
        data: Bytes,
    ) -> Pin<Box<dyn Future<Output = Result<(), StreamsClientError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn take_changelog_ts(&mut self) -> Vec<(Bytes, Option<Bytes>, Option<i64>)> { ... }
    fn apply_changelog_ts<'life0, 'async_trait>(
        &'life0 mut self,
        key: Bytes,
        value: Option<Bytes>,
        _timestamp: i64,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn as_iq(&self) -> Option<&dyn IqQueryable> { ... }
    fn set_record_context(&mut self, _ctx: RecordContext) { ... }
    fn enable_cache_erased(&mut self, _cache: Arc<Mutex<NamedCache>>) -> bool { ... }
    fn is_cached_erased(&self) -> bool { ... }
    fn flush_cache_into<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 mut self,
        _buffer: &'life1 mut VecDeque<(usize, ErasedRecord)>,
        _children: &'life2 [usize],
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
}

Lifecycle, identity, and changelog hooks for any store.

Required Methods

fn name(&self) -> &str

fn flush<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Flush pending state. This is a no-op for in-memory, where the changelog gives durability.

fn close(&mut self)

fn as_any_mut(&mut self) -> &mut dyn Any

Typed downcast hook that get_state_store uses.

fn changelog_topic(&self) -> &str

The store’s changelog topic (<app>-<store>-changelog).

fn take_changelog(&mut self) -> Vec<(Bytes, Option<Bytes>)>

Drain the buffered changelog entries: key bytes, and value bytes or None for a tombstone.

fn apply_changelog<'life0, 'async_trait>( &'life0 mut self, key: Bytes, value: Option<Bytes>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Apply a changelog record during restore. It updates state and does NOT re-log.

fn set_logging(&mut self, on: bool)

Toggle changelog logging. It is off during restore and on during processing.

fn clear<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Wipe every entry and any buffered changelog. The EOS rollback path uses this method to reset the store to a clean slate before it re-restores from the committed changelog.

fn snapshot<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Bytes> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Serialize the whole store into one barrier-snapshot payload.

The task calls this method for every store it owns when a barrier fires, and it puts the payloads into the frozen snapshot container under the store names. See crate::store::snapshot.

The payload holds every piece of state that restore_snapshot needs to rebuild the store, which for some stores is more than the key-value entries. A suppress buffer also carries its sequence counter and its byte-size total.

The payload format is Rust’s own. The container around it is shared with krabka-streams-java and krabka-streams-go, the payload inside it is not.

fn restore_snapshot<'life0, 'async_trait>( &'life0 mut self, data: Bytes, ) -> Pin<Box<dyn Future<Output = Result<(), StreamsClientError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Replace the whole store with a payload that snapshot produced.

The store is wiped first, so the restored state is the snapshot and nothing else. This method does NOT re-log, exactly as apply_changelog does not.

Errors

Returns StreamsClientError::Snapshot when the payload does not match the format this store writes.

Provided Methods

fn take_changelog_ts(&mut self) -> Vec<(Bytes, Option<Bytes>, Option<i64>)>

Like take_changelog, but each entry also carries an optional explicit changelog RECORD timestamp. None means use the producer default, which is send-time. Versioned stores override this method to emit the version timestamp (KIP-889). The default wraps take_changelog with None.

fn apply_changelog_ts<'life0, 'async_trait>( &'life0 mut self, key: Bytes, value: Option<Bytes>, _timestamp: i64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Like apply_changelog, but it carries the changelog record’s timestamp. The default ignores the timestamp and delegates. Versioned stores override this method to insert the version at this timestamp.

fn as_iq(&self) -> Option<&dyn IqQueryable>

IQ read view, if this store is interactively queryable. Default None.

fn set_record_context(&mut self, _ctx: RecordContext)

Stash the current record’s context, so a caching store can attach it to the next write. The flush forwards that context with the deduped Change. The default is a no-op.

fn enable_cache_erased(&mut self, _cache: Arc<Mutex<NamedCache>>) -> bool

Erased record-cache hook.

This method wraps the store’s backend in the supplied internal NamedCache, which is registered in the task’s ThreadCache. It returns true when this store kind is cache-aware. It lets instantiate enable caching on a materialized KV store and never learn the store’s K or V. The default is false, which means not cacheable. Window and session stores keep the default until their caching lands. KV stores override this method and delegate to their typed enable_cache.

NamedCache is crate-internal store plumbing. This method is reachable on the pub trait, but no caller outside the crate can use it.

fn is_cached_erased(&self) -> bool

Erased query for whether this store’s record cache is enabled.

The cache is enabled when enable_cache_erased has wrapped the backend. A materializing processor uses this method to decide whether to suppress its immediate downstream forward, and it never learns K or V. The cache flush forwards the deduped Change. The default is false, which means not cached and not cache-aware. KV stores override this method.

fn flush_cache_into<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, _buffer: &'life1 mut VecDeque<(usize, ErasedRecord)>, _children: &'life2 [usize], ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Flush this store’s record cache, if it has one.

This method writes the dirty entries through to the underlying store, buffers their changelog records, and pushes the deduped downstream Record<K, Change<V>> into buffer. It pushes one boxed copy PER child in children, which matches the per-child clone in ProcessorContext::forward. The default has no cache, so it is a no-op.

ErasedRecord is crate-internal graph plumbing. This method is reachable on the pub trait, but no caller outside the crate can use it.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<K: 'static, V: 'static> StateStore for JoinWindowBytesStore<K, V>

impl<K: 'static, V: 'static> StateStore for SuppressBytesStore<K, V>

impl<K: Send + 'static, V: Send + 'static> StateStore for KeyValueBytesStore<K, V>

impl<K: Send + 'static, V: Send + 'static> StateStore for SessionBytesStore<K, V>

impl<K: Send + 'static, V: Send + 'static> StateStore for VersionedBytesStore<K, V>

impl<K: Send + 'static, V: Send + 'static> StateStore for WindowBytesStore<K, V>