Struct fluvio::metadata::store::LocalStore

pub struct LocalStore<S, C> where
    C: MetadataItem,
    S: Spec,  { /* fields omitted */ }

Idempotent local memory cache of meta objects. There are only 2 write operations are permitted: sync and apply changes which are idempotent. For read, read guards are provided which provide hash map API using deref.
Hash values are wrapped in EpochCounter. EpochCounter is also deref. Using async lock to ensure read/write are thread safe.

Implementations

impl<S, C> LocalStore<S, C> where
    C: MetadataItem,
    S: Spec, 

pub fn bulk_new<N>(objects: Vec<N, Global>) -> LocalStore<S, C> where
    N: Into<MetadataStoreObject<S, C>>, 

initialize local stores with list of metadata objects

pub fn new_shared() -> Arc<LocalStore<S, C>>

create arc wrapper

pub async fn read<'a>(
    &'_ self
) -> RwLockReadGuard<'_, DualEpochMap<<S as Spec>::IndexKey, MetadataStoreObject<S, C>>>

Read guard

pub async fn write<'a>(
    &'_ self
) -> RwLockWriteGuard<'_, DualEpochMap<<S as Spec>::IndexKey, MetadataStoreObject<S, C>>>

write guard, this is private, use sync API to make changes

pub async fn epoch(&'_ self) -> i64

current epoch

pub fn init_epoch(&self) -> DualEpochCounter<()>

initial epoch that should be used store will always have epoch greater than init_epoch if there are any changes

pub async fn value<K>(
    &'_ self,
    key: &'_ K
) -> Option<DualEpochCounter<MetadataStoreObject<S, C>>> where
    K: Eq + Hash + ?Sized,
    <S as Spec>::IndexKey: Borrow<K>, 

copy of the value

pub async fn spec<K>(&'_ self, key: &'_ K) -> Option<S> where
    K: Eq + Hash + ?Sized,
    <S as Spec>::IndexKey: Borrow<K>, 

copy spec

pub async fn find_and_do<K, F>(&'_ self, key: &'_ K, func: F) -> Option<()> where
    K: Eq + Hash,
    F: FnMut(&MetadataStoreObject<S, C>),
    <S as Spec>::IndexKey: Borrow<K>, 

iterate over entry

pub async fn contains_key<K>(&'_ self, key: &'_ K) -> bool where
    K: Eq + Hash + ?Sized,
    <S as Spec>::IndexKey: Borrow<K>, 

pub async fn count(&'_ self) -> i32

pub async fn clone_specs(&'_ self) -> Vec<S, Global>

pub async fn clone_keys(&'_ self) -> Vec<<S as Spec>::IndexKey, Global>

pub async fn clone_values(&'_ self) -> Vec<MetadataStoreObject<S, C>, Global>

pub fn event_publisher(&self) -> &EventPublisher

pub fn change_listener(self: &Arc<LocalStore<S, C>>) -> ChangeListener<S, C>

create new change listener

impl<S, C> LocalStore<S, C> where
    C: MetadataItem + PartialEq<C>,
    S: Spec + PartialEq<S>,
    <S as Spec>::Status: PartialEq<<S as Spec>::Status>,
    <S as Spec>::IndexKey: Display, 

pub async fn sync_all(
    &'_ self,
    incoming_changes: Vec<MetadataStoreObject<S, C>, Global>
) -> SyncStatus

sync with incoming changes as source of truth. any objects not in incoming list will be deleted after sync operation, prior history will be removed and any subsequent change query will return full list instead of changes

pub async fn apply_changes(
    &'_ self,
    changes: Vec<LSUpdate<S, C>, Global>
) -> Option<SyncStatus>

apply changes to this store if item doesn't exit, it will be treated as add if item exist but different, it will be treated as updated epoch will be only incremented if there are actual changes which means this is idempotent operations. same add result in only 1 single epoch increase.

Trait Implementations

impl<S, C> Debug for LocalStore<S, C> where
    C: Debug + MetadataItem,
    S: Debug + Spec,
    <S as Spec>::IndexKey: Debug, 

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

Formats the value using the given formatter.

impl<S, C> Default for LocalStore<S, C> where
    C: MetadataItem,
    S: Spec, 

pub fn default() -> LocalStore<S, C>

Returns the "default value" for a type.

impl<S, C> Display for LocalStore<S, C> where
    C: MetadataItem,
    S: Spec, 

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

Formats the value using the given formatter.

impl<C> PartitionLocalStorePolicy<C> for LocalStore<PartitionSpec, C> where
    C: MetadataItem + Send + Sync, 

pub fn names<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<ReplicaKey, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

pub fn topic_partitions<'life0, 'life1, 'async_trait>(
    &'life0 self,
    topic: &'life1 str
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<PartitionSpec, C>, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    'life1: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

pub fn partition_spec_for_spu<'life0, 'async_trait>(
    &'life0 self,
    target_spu: i32
) -> Pin<Box<dyn Future<Output = Vec<(ReplicaKey, PartitionSpec), Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

find all partitions that has spu in the replicas

pub fn count_topic_partitions<'life0, 'life1, 'async_trait>(
    &'life0 self,
    topic: &'life1 str
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    'life1: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

pub fn topic_partitions_list<'life0, 'life1, 'async_trait>(
    &'life0 self,
    topic: &'life1 str
) -> Pin<Box<dyn Future<Output = Vec<ReplicaKey, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    'life1: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

pub fn table_fmt<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

pub fn replica_for_spu<'life0, 'async_trait>(
    &'life0 self,
    target_spu: i32
) -> Pin<Box<dyn Future<Output = Vec<Replica, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

replica msg for target spu

pub fn leaders<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<ReplicaLeader, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<PartitionSpec, C>: 'async_trait, 

pub fn bulk_load<S>(
    partitions: Vec<((S, i32), Vec<i32, Global>), Global>
) -> LocalStore<PartitionSpec, C> where
    S: Into<String>, 

impl<C> SpuLocalStorePolicy<C> for LocalStore<SpuSpec, C> where
    C: MetadataItem + Send + Sync, 

pub fn online_status<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = HashSet<i32, RandomState>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn online_spu_count<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

count online SPUs

pub fn spu_used_for_replica<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

count spus that can be used for replica

pub fn online_spu_ids<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<i32, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn spu_ids<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<i32, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn online_spus<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<SpuSpec, C>, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn custom_spus<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<SpuSpec, C>, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn get_by_id<'life0, 'async_trait>(
    &'life0 self,
    id: i32
) -> Pin<Box<dyn Future<Output = Option<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn validate_spu_for_registered<'life0, 'async_trait>(
    &'life0 self,
    id: i32
) -> Pin<Box<dyn Future<Output = bool> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn all_names<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<String, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn table_fmt<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn spus_in_rack_count<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

number of spus in rack count

pub fn live_spu_rack_map_sorted<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<(String, Vec<i32, Global>), Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn online_spu_rack_map<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = BTreeMap<String, Vec<i32, Global>>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn online_spus_in_rack(
    rack_map: &[(String, Vec<i32, Global>)]
) -> Vec<i32, Global>

pub fn all_spus_to_spu_msgs<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Vec<Message<SpuSpec>, Global>> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<SpuSpec, C>: 'async_trait, 

pub fn quick(
    spus: Vec<(i32, bool, Option<String>), Global>
) -> LocalStore<SpuSpec, C>

impl<C> TopicLocalStorePolicy<C> for LocalStore<TopicSpec, C> where
    C: MetadataItem + Send + Sync, 

pub fn table_fmt<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send, Global>> where
    'life0: 'async_trait,
    LocalStore<TopicSpec, C>: 'async_trait,