qdb 0.1.10

Library for interacting with QDB
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
114
115

use crate::framework::application::Context;
use crate::framework::workers::common::WorkerTrait;
use crate::framework::events::emitter::Emitter;

use crate::Result;

use std::sync::mpsc::Receiver;

pub struct Emitters {
    pub connection_status: Emitter<bool>,
}

pub struct Receivers {
    pub network_connection_status: Option<Receiver<bool>>,
}

pub struct Worker {
    is_db_connected: bool,
    is_nw_connected: bool,
    pub emitters: Emitters,
    pub receivers: Receivers,
}

impl Worker {
    pub fn new() -> Self {
        Self {
            is_db_connected: false,
            is_nw_connected: false,
            emitters: Emitters {
                connection_status: Emitter::new(),
            },
            receivers: Receivers {
                network_connection_status: None,
            },
        }
    }
}

impl WorkerTrait for Worker {
    fn intialize(&mut self, ctx: Context) -> Result<()> {
        let c = format!("{}::{}", std::any::type_name::<Self>(), "initialize");

        ctx.logger().info(
            format!("[{}] Initializing database worker", c).as_str(),
        );
        Ok(())
    }

    fn do_work(&mut self, ctx: Context) -> Result<()> {
        let c = format!("{}::{}", std::any::type_name::<Self>(), "do_work");

        if !self.is_nw_connected {
            if self.is_db_connected {
                ctx.logger().warning(
                    format!("[{}] Network connection loss has disrupted database connection", c).as_str()
                );
                self.is_db_connected = false;
                self.emitters.connection_status.emit(self.is_db_connected);
            }

            return Ok(());
        }

        if !ctx.database().connected() {
            if self.is_db_connected {
                ctx.logger().warning(
                    format!("[{}] Disconnected from database", c).as_str(),
                );
                ctx.database().clear_notifications();
                self.is_db_connected = false;
                self.emitters.connection_status.emit(self.is_db_connected);
            }

            ctx.logger().debug(
                format!("[{}] Attempting to connect to the database...", c).as_str(),
            );

            ctx.database().disconnect();
            ctx.database().connect()?;

            if ctx.database().connected() {
                ctx.logger().info(
                    format!("[{}] Connected to the database", c).as_str(),
                );
                self.is_db_connected = true;
                self.emitters.connection_status.emit(self.is_db_connected);
            }

            return Ok(());
        }

        ctx.database().process_notifications()?;

        Ok(())
    }

    fn deinitialize(&mut self, ctx: Context) -> Result<()> {
        let c = format!("{}::{}", std::any::type_name::<Self>(), "deinitialize");

        ctx.logger().info(
            format!("[{}] Deinitializing database worker", c).as_str(),
        );
        Ok(())
    }

    fn process_events(&mut self) -> Result<()> {
        if let Some(receiver) = &self.receivers.network_connection_status {
            while let Ok(connected) = receiver.try_recv() {
                self.is_nw_connected = connected;
            }
        }

        Ok(())
    }
}