pito_sync/
runner.rs

1use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
2use std::thread;
3use std::time::Duration;
4
5use rusqlite::Connection;
6
7use crate::book::now_ms;
8use crate::{Failure, Secret, Spec, due, failed, open, retry_after, started};
9
10const MINUTE: Duration = Duration::from_secs(60);
11const HOUR: Duration = Duration::from_secs(3600);
12
13#[derive(Debug, Clone, PartialEq, Eq, Hash)]
14pub struct Changed(pub String);
15
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct Resync(pub String, pub Option<String>);
18
19#[derive(Debug)]
20pub enum Message {
21    Resync(Resync),
22    Secret(String, Secret),
23}
24
25impl From<Resync> for Message {
26    fn from(resync: Resync) -> Self {
27        Message::Resync(resync)
28    }
29}
30
31pub trait Source: Send {
32    fn name(&self) -> &str;
33    fn every(&self) -> Option<Duration>;
34    fn pass(&mut self, conn: &mut Connection, id: Option<&str>) -> Result<usize, Failure>;
35    fn secret(&mut self, name: &str, value: Secret) {
36        let _ = (name, value);
37    }
38}
39
40pub fn spawn(spec: &Spec, source: Box<dyn Source>, changed: Sender<Changed>) -> Sender<Message> {
41    let spec = spec.clone();
42    let (asks, inbox) = mpsc::channel::<Message>();
43    thread::spawn(move || match open(&spec) {
44        Ok(conn) => run(conn, source, &inbox, &changed),
45        Err(_) => {
46            let _ = changed.send(Changed(source.name().to_owned()));
47        }
48    });
49    asks
50}
51
52pub fn spawn_checked(
53    spec: &Spec,
54    source: Box<dyn Source>,
55    changed: Sender<Changed>,
56) -> Result<Sender<Message>, String> {
57    let conn = open(spec)?;
58    let (asks, inbox) = mpsc::channel::<Message>();
59    thread::spawn(move || run(conn, source, &inbox, &changed));
60    Ok(asks)
61}
62
63pub fn pass_once(
64    conn: &mut Connection,
65    source: &mut dyn Source,
66    id: Option<&str>,
67) -> Result<usize, Failure> {
68    pass_announced(conn, source, id, || true).0
69}
70
71fn pass_announced(
72    conn: &mut Connection,
73    source: &mut dyn Source,
74    id: Option<&str>,
75    announce: impl FnOnce() -> bool,
76) -> (Result<usize, Failure>, bool) {
77    let name = source.name().to_owned();
78    let _ = started(conn, &name, now_ms());
79    if !announce() {
80        return (Ok(0), false);
81    }
82    let outcome = source.pass(conn, id);
83    if let Err(failure) = &outcome {
84        let _ = failed(conn, &name, failure);
85    }
86    (outcome, true)
87}
88
89fn once(
90    conn: &mut Connection,
91    source: &mut dyn Source,
92    name: &str,
93    id: Option<&str>,
94    changed: &Sender<Changed>,
95) -> Option<()> {
96    let (_, ran) = pass_announced(conn, source, id, || {
97        changed.send(Changed(name.to_owned())).is_ok()
98    });
99    if !ran {
100        return None;
101    }
102    changed.send(Changed(name.to_owned())).ok()
103}
104
105fn run(
106    mut conn: Connection,
107    mut source: Box<dyn Source>,
108    inbox: &Receiver<Message>,
109    changed: &Sender<Changed>,
110) {
111    let name = source.name().to_owned();
112    let wait = source.every().unwrap_or(HOUR).min(MINUTE);
113    let held = retry_after(&conn, &name)
114        .ok()
115        .flatten()
116        .is_some_and(|at| now_ms() < at);
117    if !held && once(&mut conn, source.as_mut(), &name, None, changed).is_none() {
118        return;
119    }
120    loop {
121        let id = match inbox.recv_timeout(wait) {
122            Ok(Message::Resync(Resync(_, id))) => id,
123            Ok(Message::Secret(key, value)) => {
124                source.secret(&key, value);
125                None
126            }
127            Err(RecvTimeoutError::Timeout) => {
128                if !source
129                    .every()
130                    .is_some_and(|every| due(&conn, &name, every, now_ms()))
131                {
132                    continue;
133                }
134                None
135            }
136            Err(RecvTimeoutError::Disconnected) => return,
137        };
138        if once(&mut conn, source.as_mut(), &name, id.as_deref(), changed).is_none() {
139            return;
140        }
141    }
142}
pito_sync/runner.rs

pito_sync/
runner.rs
