[−]Struct fluvio::metadata::store::LocalStore
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,
C: MetadataItem,
S: Spec,
pub fn bulk_new<N>(objects: Vec<N>) -> LocalStore<S, C> where
N: Into<MetadataStoreObject<S, C>>,
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<'_, EpochMap<<S as Spec>::IndexKey, MetadataStoreObject<S, C>>>
&'_ self
) -> RwLockReadGuard<'_, EpochMap<<S as Spec>::IndexKey, MetadataStoreObject<S, C>>>
Read guard
pub async fn write<'a>(
&'_ self
) -> RwLockWriteGuard<'_, EpochMap<<S as Spec>::IndexKey, MetadataStoreObject<S, C>>>
&'_ self
) -> RwLockWriteGuard<'_, EpochMap<<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) -> EpochCounter<()>
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<EpochCounter<MetadataStoreObject<S, C>>> where
K: Eq + Hash + ?Sized,
<S as Spec>::IndexKey: Borrow<K>,
&'_ self,
key: &'_ K
) -> Option<EpochCounter<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>,
K: Eq + Hash + ?Sized,
<S as Spec>::IndexKey: Borrow<K>,
copy spec
pub async fn find_and_do<K, F>(&'_ self, key: &'_ K, __arg2: F) -> Option<()> where
F: FnMut(&MetadataStoreObject<S, C>),
K: Eq + Hash,
<S as Spec>::IndexKey: Borrow<K>,
F: FnMut(&MetadataStoreObject<S, C>),
K: Eq + Hash,
<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>,
K: Eq + Hash + ?Sized,
<S as Spec>::IndexKey: Borrow<K>,
pub async fn count(&'_ self) -> i32
pub async fn clone_specs(&'_ self) -> Vec<S>
pub async fn clone_keys(&'_ self) -> Vec<<S as Spec>::IndexKey>
pub async fn clone_values(&'_ self) -> Vec<MetadataStoreObject<S, C>>
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,
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>>
) -> SyncStatus
&'_ self,
incoming_changes: Vec<MetadataStoreObject<S, C>>
) -> SyncStatus
sync list of changes with with this store assuming incoming is source of truth. any objects not in incoming list will be deleted (but will be in history). 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>>
) -> Option<SyncStatus>
&'_ self,
changes: Vec<LSUpdate<S, C>>
) -> Option<SyncStatus>
apply changes to this store this assume changes are source of truth. 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,
C: Debug + MetadataItem,
S: Debug + Spec,
<S as Spec>::IndexKey: Debug,
impl<S, C> Default for LocalStore<S, C> where
C: MetadataItem,
S: Spec,
C: MetadataItem,
S: Spec,
fn default() -> LocalStore<S, C>
impl<S, C> Display for LocalStore<S, C> where
C: MetadataItem,
S: Spec,
C: MetadataItem,
S: Spec,
impl<C> PartitionLocalStorePolicy<C> for LocalStore<PartitionSpec, C> where
C: MetadataItem + Send + Sync,
C: MetadataItem + Send + Sync,
fn names<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<ReplicaKey>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<ReplicaKey>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
fn topic_partitions<'life0, 'life1, 'async_trait>(
&'life0 self,
topic: &'life1 str
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<PartitionSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
'life1: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self,
topic: &'life1 str
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<PartitionSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
'life1: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
fn partition_spec_for_spu<'life0, 'async_trait>(
&'life0 self,
target_spu: i32
) -> Pin<Box<dyn Future<Output = Vec<(ReplicaKey, PartitionSpec)>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self,
target_spu: i32
) -> Pin<Box<dyn Future<Output = Vec<(ReplicaKey, PartitionSpec)>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
find all partitions that has spu in the replicas
fn count_topic_partitions<'life0, 'life1, 'async_trait>(
&'life0 self,
topic: &'life1 str
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
'life1: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self,
topic: &'life1 str
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
'life1: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
fn topic_partitions_list<'life0, 'life1, 'async_trait>(
&'life0 self,
topic: &'life1 str
) -> Pin<Box<dyn Future<Output = Vec<ReplicaKey>> + 'async_trait + Send>> where
'life0: 'async_trait,
'life1: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self,
topic: &'life1 str
) -> Pin<Box<dyn Future<Output = Vec<ReplicaKey>> + 'async_trait + Send>> where
'life0: 'async_trait,
'life1: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
fn table_fmt<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
fn replica_for_spu<'life0, 'async_trait>(
&'life0 self,
target_spu: i32
) -> Pin<Box<dyn Future<Output = Vec<Replica>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self,
target_spu: i32
) -> Pin<Box<dyn Future<Output = Vec<Replica>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
replica msg for target spu
fn leaders<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<ReplicaLeader>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<ReplicaLeader>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<PartitionSpec, C>: 'async_trait,
fn bulk_load<S>(
partitions: Vec<((S, i32), Vec<i32>)>
) -> LocalStore<PartitionSpec, C> where
S: Into<String>,
partitions: Vec<((S, i32), Vec<i32>)>
) -> LocalStore<PartitionSpec, C> where
S: Into<String>,
impl<C> SpuLocalStorePolicy<C> for LocalStore<SpuSpec, C> where
C: MetadataItem + Send + Sync,
C: MetadataItem + Send + Sync,
fn online_status<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = HashSet<i32, RandomState>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = HashSet<i32, RandomState>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn online_spu_count<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
count online SPUs
fn spu_used_for_replica<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
count spus that can be used for replica
fn online_spu_ids<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<i32>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<i32>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn spu_ids<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<i32>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<i32>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn online_spus<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn custom_spus<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn get_by_id<'life0, 'async_trait>(
&'life0 self,
id: i32
) -> Pin<Box<dyn Future<Output = Option<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self,
id: i32
) -> Pin<Box<dyn Future<Output = Option<MetadataStoreObject<SpuSpec, C>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn validate_spu_for_registered<'life0, 'async_trait>(
&'life0 self,
id: i32
) -> Pin<Box<dyn Future<Output = bool> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self,
id: i32
) -> Pin<Box<dyn Future<Output = bool> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn all_names<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<String>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<String>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn table_fmt<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = String> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn spus_in_rack_count<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = i32> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
number of spus in rack count
fn live_spu_rack_map_sorted<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<(String, Vec<i32>)>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<(String, Vec<i32>)>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn online_spu_rack_map<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = BTreeMap<String, Vec<i32>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = BTreeMap<String, Vec<i32>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn online_spus_in_rack(rack_map: &[(String, Vec<i32>)]) -> Vec<i32>
fn all_spus_to_spu_msgs<'life0, 'async_trait>(
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<Message<SpuSpec>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
&'life0 self
) -> Pin<Box<dyn Future<Output = Vec<Message<SpuSpec>>> + 'async_trait + Send>> where
'life0: 'async_trait,
LocalStore<SpuSpec, C>: 'async_trait,
fn quick(spus: Vec<(i32, bool, Option<String>)>) -> LocalStore<SpuSpec, C>
impl<C> TopicLocalStorePolicy<C> for LocalStore<TopicSpec, C> where
C: MetadataItem + Send + Sync,
C: MetadataItem + Send + Sync,
Auto Trait Implementations
impl<S, C> !RefUnwindSafe for LocalStore<S, C>
impl<S, C> Send for LocalStore<S, C> where
C: Send,
S: Send,
<S as Spec>::IndexKey: Send,
<S as Spec>::Status: Send,
C: Send,
S: Send,
<S as Spec>::IndexKey: Send,
<S as Spec>::Status: Send,
impl<S, C> Sync for LocalStore<S, C> where
C: Send + Sync,
S: Send + Sync,
<S as Spec>::IndexKey: Send + Sync,
<S as Spec>::Status: Send + Sync,
C: Send + Sync,
S: Send + Sync,
<S as Spec>::IndexKey: Send + Sync,
<S as Spec>::Status: Send + Sync,
impl<S, C> Unpin for LocalStore<S, C> where
C: Unpin,
S: Unpin,
<S as Spec>::IndexKey: Unpin,
<S as Spec>::Status: Unpin,
C: Unpin,
S: Unpin,
<S as Spec>::IndexKey: Unpin,
<S as Spec>::Status: Unpin,
impl<S, C> UnwindSafe for LocalStore<S, C> where
C: UnwindSafe,
S: UnwindSafe,
<S as Spec>::IndexKey: UnwindSafe,
<S as Spec>::Status: UnwindSafe,
C: UnwindSafe,
S: UnwindSafe,
<S as Spec>::IndexKey: UnwindSafe,
<S as Spec>::Status: UnwindSafe,
Blanket Implementations
impl<T> Any for T where
T: 'static + ?Sized,
[src]
T: 'static + ?Sized,
impl<T> Borrow<T> for T where
T: ?Sized,
[src]
T: ?Sized,
impl<T> BorrowMut<T> for T where
T: ?Sized,
[src]
T: ?Sized,
fn borrow_mut(&mut self) -> &mut T
[src]
impl<T> Erased for T
impl<T> From<T> for T
[src]
impl<T> Instrument for T
[src]
fn instrument(self, span: Span) -> Instrumented<Self>
[src]
fn in_current_span(self) -> Instrumented<Self>
[src]
impl<T> Instrument for T
[src]
fn instrument(self, span: Span) -> Instrumented<Self>
[src]
fn in_current_span(self) -> Instrumented<Self>
[src]
impl<T, U> Into<U> for T where
U: From<T>,
[src]
U: From<T>,
impl<T> ToString for T where
T: Display + ?Sized,
[src]
T: Display + ?Sized,
impl<T, U> TryFrom<U> for T where
U: Into<T>,
[src]
U: Into<T>,
type Error = Infallible
The type returned in the event of a conversion error.
fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>
[src]
impl<T, U> TryInto<U> for T where
U: TryFrom<T>,
[src]
U: TryFrom<T>,
type Error = <U as TryFrom<T>>::Error
The type returned in the event of a conversion error.
fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>
[src]
impl<T> WithSubscriber for T
[src]
fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self> where
S: Into<Dispatch>,
[src]
S: Into<Dispatch>,