rlink 0.6.16

High performance Stream Processing Framework
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

use crate::core::backend::CheckpointBackend;
use crate::core::checkpoint::Checkpoint;
use crate::core::runtime::CheckpointId;
use crate::storage::checkpoint::memory_checkpoint_storage::MemoryCheckpointStorage;
use crate::storage::checkpoint::mysql_checkpoint_storage::MySqlCheckpointStorage;

pub mod memory_checkpoint_storage;
pub mod mysql_checkpoint_storage;

pub trait TCheckpointStorage {
    fn save(
        &mut self,
        application_name: &str,
        application_id: &str,
        checkpoint_id: CheckpointId,
        finish_cks: Vec<Checkpoint>,
        ttl: u64,
    ) -> anyhow::Result<()>;

    fn load(
        &mut self,
        application_name: &str,
        application_id: &str,
    ) -> anyhow::Result<Vec<Checkpoint>>;

    fn load_by_checkpoint_id(
        &mut self,
        application_name: &str,
        application_id: &str,
        checkpoint_id: CheckpointId,
    ) -> anyhow::Result<Vec<Checkpoint>>;
}

pub enum CheckpointStorage {
    MemoryCheckpointStorage(MemoryCheckpointStorage),
    MySqlCheckpointStorage(MySqlCheckpointStorage),
}

impl CheckpointStorage {
    pub fn new(checkpoint_backend: &CheckpointBackend) -> Self {
        match checkpoint_backend {
            CheckpointBackend::Memory => {
                CheckpointStorage::MemoryCheckpointStorage(MemoryCheckpointStorage::new())
            }
            CheckpointBackend::MySql { endpoint, table } => {
                CheckpointStorage::MySqlCheckpointStorage(MySqlCheckpointStorage::new(
                    endpoint.clone(),
                    table.clone(),
                ))
            }
        }
    }
}

impl TCheckpointStorage for CheckpointStorage {
    fn save(
        &mut self,
        application_name: &str,
        application_id: &str,
        checkpoint_id: CheckpointId,
        finish_cks: Vec<Checkpoint>,
        ttl: u64,
    ) -> anyhow::Result<()> {
        match self {
            CheckpointStorage::MemoryCheckpointStorage(storage) => storage.save(
                application_name,
                application_id,
                checkpoint_id,
                finish_cks,
                ttl,
            ),
            CheckpointStorage::MySqlCheckpointStorage(storage) => storage.save(
                application_name,
                application_id,
                checkpoint_id,
                finish_cks,
                ttl,
            ),
        }
    }

    fn load(
        &mut self,
        application_name: &str,
        application_id: &str,
    ) -> anyhow::Result<Vec<Checkpoint>> {
        match self {
            CheckpointStorage::MemoryCheckpointStorage(storage) => {
                storage.load(application_name, application_id)
            }
            CheckpointStorage::MySqlCheckpointStorage(storage) => {
                storage.load(application_name, application_id)
            }
        }
    }

    fn load_by_checkpoint_id(
        &mut self,
        application_name: &str,
        application_id: &str,
        checkpoint_id: CheckpointId,
    ) -> anyhow::Result<Vec<Checkpoint>> {
        match self {
            CheckpointStorage::MemoryCheckpointStorage(storage) => {
                storage.load_by_checkpoint_id(application_name, application_id, checkpoint_id)
            }
            CheckpointStorage::MySqlCheckpointStorage(storage) => {
                storage.load_by_checkpoint_id(application_name, application_id, checkpoint_id)
            }
        }
    }
}