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

use crate::Error;
use axum::response::sse;
use cryptr::utils::b64_encode;
use tokio::task;
use tracing::{debug, error, info, warn};

pub enum NotifyRequest {
    Notify((i64, Vec<u8>)),
    Listen((flume::Sender<Result<sse::Event, Error>>)),
}

pub fn spawn() -> (
    flume::Sender<NotifyRequest>,
    flume::Receiver<(i64, Vec<u8>)>,
) {
    let (tx_req, rx_req) = flume::unbounded();
    let (tx_local, rx_local) = flume::unbounded();
    task::spawn(handler(rx_req, tx_local));
    (tx_req, rx_local)
}

#[tracing::instrument(level = "debug", skip_all)]
async fn handler(rx_req: flume::Receiver<NotifyRequest>, tx_local: flume::Sender<(i64, Vec<u8>)>) {
    let mut listeners: Vec<flume::Sender<Result<sse::Event, Error>>> = Vec::new();
    let mut remove_indexes = Vec::new();

    while let Ok(req) = rx_req.recv_async().await {
        match req {
            NotifyRequest::Notify((ts, data)) => {
                debug!("new notification from {}", ts);

                if !listeners.is_empty() {
                    let event = sse::Event::default().data(format!("{} {}", ts, b64_encode(&data)));

                    for (idx, listener) in listeners.iter().enumerate() {
                        // unbounded channels can never block
                        if let Err(err) = listener.send(Ok(event.clone())) {
                            error!("Error sending listener Notification: {}", err);
                            remove_indexes.push(idx);
                        }
                    }

                    while let Some(idx) = remove_indexes.pop() {
                        info!("Removing Notification Listener at position {}", idx);
                        listeners.swap_remove(idx);
                    }
                }

                // unbounded channels can never block
                if let Err(err) = tx_local.send((ts, data)) {
                    error!("Error sending local Notification: {}", err);
                    break;
                }
            }
            NotifyRequest::Listen(tx) => {
                info!("New notification listener subscribed");
                listeners.push(tx);
            }
        }
    }

    debug!("Listen / Notify handler exiting");
}