fluvio-spu 0.2.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

mod global_context;
mod store;
pub(crate) mod storage;

pub mod spus;
pub mod replica;

pub use self::global_context::GlobalContext;
pub use self::store::Spec;
pub use self::store::LocalStore;
pub use self::store::SpecChange;

pub use self::spus::SpuLocalStore;
pub use self::replica::SharedReplicaLocalStore;

use std::sync::Arc;
use ::fluvio_storage::FileReplica;
use fluvio_socket::SinkPool;
use fluvio_types::SpuId;
use crate::config::SpuConfig;

pub type SharedGlobalContext<S> = Arc<GlobalContext<S>>;
pub type DefaultSharedGlobalContext = SharedGlobalContext<FileReplica>;
pub type SharedSpuSinks = Arc<SinkPool<SpuId>>;
pub type SharedSpuConfig = Arc<SpuConfig>;

pub use event::OffsetUpdateEvent;

mod event {

    use dataplane::Offset;
    use fluvio_controlplane_metadata::partition::ReplicaKey;

    /// used for communicating change in offset for any replica
    #[derive(Debug, Clone)]
    pub struct OffsetUpdateEvent {
        pub replica_id: ReplicaKey,
        pub leo: Offset,
        pub hw: Offset,
    }
}

mod broadcast {

    pub use tokio::sync::broadcast::Receiver;
    pub use tokio::sync::broadcast::Sender;
    pub use tokio::sync::broadcast::channel;
    pub use tokio::sync::broadcast::RecvError;

    #[derive(Debug)]
    pub struct Channel<T> {
        receiver: Receiver<T>,
        sender: Sender<T>,
    }

    impl<T> Channel<T> {
        pub fn new(capacity: usize) -> Self {
            let (sender, receiver) = channel(capacity);
            Self { receiver, sender }
        }

        /// create new clone of sender
        pub fn receiver(&self) -> Receiver<T> {
            self.sender.subscribe()
        }

        pub fn sender(&self) -> Sender<T> {
            self.sender.clone()
        }
    }
}