hiqlite 0.13.0

Hiqlite - highly-available, embeddable, raft-based SQLite + cache
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

use crate::app_state::AppState;
use crate::client::stream::ClientStreamReq;
use crate::{Client, Error};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;
use tokio::time;
use tracing::{debug, info};

pub struct ShutdownHandle {
    state: Arc<AppState>,
    #[cfg(feature = "cache")]
    with_tls: bool,
    #[cfg(feature = "cache")]
    tls_no_verify: bool,
    #[cfg(feature = "cache")]
    tx_client_cache: flume::Sender<ClientStreamReq>,
    #[cfg(feature = "sqlite")]
    tx_client_db: flume::Sender<ClientStreamReq>,
    tx_shutdown: Option<watch::Sender<bool>>,
    rx_shutdown: watch::Receiver<bool>,
}

impl ShutdownHandle {
    pub async fn wait(&mut self) -> Result<(), Error> {
        let _ = self.rx_shutdown.changed().await;
        info!("ShutdownHandle received shutdown signal - shutting down the Raft node now");

        if time::timeout(
            Duration::from_secs(15),
            Client::shutdown_execute(
                &self.state,
                #[cfg(feature = "cache")]
                self.with_tls,
                #[cfg(feature = "cache")]
                self.tls_no_verify,
                #[cfg(feature = "cache")]
                &self.tx_client_cache,
                #[cfg(feature = "sqlite")]
                &self.tx_client_db,
                &self.tx_shutdown,
            ),
        )
        .await
        .is_err()
        {
            debug!("Timeout reached while waiting for Client::shutdown_execute");
        }
        Ok(())
    }
}

impl Client {
    #[allow(unused)]
    pub fn shutdown_handle(&self) -> Result<ShutdownHandle, Error> {
        if let Some(state) = self.inner.state.clone() {
            let (tx, rx_shutdown) = watch::channel(false);

            ctrlc::set_handler(move || {
                let _ = tx.send(true);
            })
            .expect("Error setting shutdown handle");

            Ok(ShutdownHandle {
                state,
                #[cfg(feature = "cache")]
                with_tls: self.inner.tls_config.is_some(),
                #[cfg(feature = "cache")]
                tls_no_verify: self.inner.tls_no_verify,
                #[cfg(feature = "cache")]
                tx_client_cache: self.inner.tx_client_cache.clone(),
                #[cfg(feature = "sqlite")]
                tx_client_db: self.inner.tx_client_db.clone(),
                tx_shutdown: self.inner.tx_shutdown.clone(),
                rx_shutdown,
            })
        } else {
            Err(Error::Error(
                "A shutdown handle can only be registered for local clients".into(),
            ))
        }
    }
}