fluvio-spu 0.3.0

Fluvio Stream Processing Unit
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

//!
//! # Global Context
//!
//! Global Context stores entities that persist through system operation.
//!
use std::sync::Arc;
use std::fmt::Debug;

use fluvio_socket::SharedSinkPool;
use fluvio_socket::SinkPool;
use fluvio_types::SpuId;
use fluvio_storage::ReplicaStorage;

use crate::core::broadcast::Channel;
use crate::config::SpuConfig;
use crate::controllers::leader_replica::SharedReplicaLeadersState;
use crate::controllers::follower_replica::FollowersState;
use crate::controllers::follower_replica::SharedFollowersState;
use crate::controllers::leader_replica::ReplicaLeadersState;
use super::spus::SharedSpuLocalStore;
use super::SharedReplicaLocalStore;
use super::spus::SpuLocalStore;
use super::replica::ReplicaStore;
use super::SharedSpuConfig;
use super::OffsetUpdateEvent;

#[derive(Debug)]
pub struct GlobalContext<S> {
    config: SharedSpuConfig,
    spu_localstore: SharedSpuLocalStore,
    replica_localstore: SharedReplicaLocalStore,
    leaders_state: SharedReplicaLeadersState<S>,
    followers_state: SharedFollowersState<S>,
    follower_sinks: SharedSinkPool<SpuId>,
    offset_channel: Channel<OffsetUpdateEvent>,
}

// -----------------------------------
// Global Contesxt - Implementation
// -----------------------------------

impl<S> GlobalContext<S>
where
    S: ReplicaStorage + Debug,
{
    pub fn new_shared_context(spu_config: SpuConfig) -> Arc<Self> {
        Arc::new(GlobalContext::new(spu_config))
    }

    pub fn new(spu_config: SpuConfig) -> Self {
        GlobalContext {
            spu_localstore: SpuLocalStore::new_shared(),
            replica_localstore: ReplicaStore::new_shared(),
            config: Arc::new(spu_config),
            follower_sinks: SinkPool::new_shared(),
            leaders_state: ReplicaLeadersState::new_shared(),
            followers_state: FollowersState::new_shared(),
            offset_channel: Channel::new(100),
        }
    }

    pub fn spu_localstore_owned(&self) -> SharedSpuLocalStore {
        self.spu_localstore.clone()
    }

    /// retrieves local spu id
    pub fn local_spu_id(&self) -> SpuId {
        self.config.id
    }

    pub fn spu_localstore(&self) -> &SpuLocalStore {
        &self.spu_localstore
    }

    pub fn replica_localstore(&self) -> &ReplicaStore {
        &self.replica_localstore
    }
    pub fn follower_sinks(&self) -> &SinkPool<SpuId> {
        &self.follower_sinks
    }

    pub fn followers_sink_owned(&self) -> SharedSinkPool<SpuId> {
        self.follower_sinks.clone()
    }

    pub fn leaders_state(&self) -> &ReplicaLeadersState<S> {
        &self.leaders_state
    }

    pub fn leader_state_owned(&self) -> SharedReplicaLeadersState<S> {
        self.leaders_state.clone()
    }

    pub fn followers_state(&self) -> &FollowersState<S> {
        &self.followers_state
    }

    pub fn followers_state_owned(&self) -> SharedFollowersState<S> {
        self.followers_state.clone()
    }

    pub fn config(&self) -> &SpuConfig {
        &self.config
    }

    pub fn config_owned(&self) -> SharedSpuConfig {
        self.config.clone()
    }

    pub fn offset_channel(&self) -> &Channel<OffsetUpdateEvent> {
        &self.offset_channel
    }
}