Struct WindowBytesStore 

pub struct WindowBytesStore<K, V> { /* private fields */ }

Implementations

impl<K: 'static, V: 'static> WindowBytesStore<K, V>

pub fn in_memory( name: String, key_serde: Box<dyn Serde<K>>, value_serde: Box<dyn Serde<V>>, changelog_topic: String, window_size: Time, ) -> Self

Trait Implementations

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

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 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.

fn is_cached_erased(&self) -> bool

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

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.

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.

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.

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.

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

fn fetch_single<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 K, window_start: i64, ) -> Pin<Box<dyn Future<Output = Option<(i64, V)>> + Send + 'async_trait>>

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

fn fetch<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 K, time_from: i64, time_to: i64, ) -> Pin<Box<dyn Future<Output = Vec<(i64, V)>> + Send + 'async_trait>>

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

fn fetch_with_ts<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 K, time_from: i64, time_to: i64, ) -> Pin<Box<dyn Future<Output = Vec<(i64, i64, V)>> + Send + 'async_trait>>

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

Like fetch, but also returns each window’s stored record timestamp.

fn fetch_all_in_range<'life0, 'async_trait>( &'life0 self, start_from: i64, start_to: i64, ) -> Pin<Box<dyn Future<Output = Vec<(K, i64, i64, V)>> + Send + 'async_trait>>

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

Returns every window across ALL keys whose windowStart is in [start_from, start_to].

fn put<'life0, 'async_trait>( &'life0 mut self, key: K, window_start: i64, value: V, record_ts: i64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

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