sofie_db/worker/
feeds.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use sofie_core::{Error, Result};
5use tokio::sync::watch;
6use tokio::task::JoinHandle;
7
8use super::{Registry, Worker};
9use crate::{Db, Entry, FeedSource, Trigger};
10
11const PRUNE: &str = "sofie.feed.prune";
12
13#[derive(Default)]
14pub struct Feeds {
15    published: Vec<&'static str>,
16    pulled: Vec<Arc<dyn FeedSource>>,
17}
18
19impl Feeds {
20    /// Old entries are pruned hourly.
21    #[must_use]
22    pub fn publish(mut self, event: &'static str) -> Self {
23        self.published.push(event);
24        self
25    }
26
27    /// Each pulled event is delivered once.
28    #[must_use]
29    pub fn pull(mut self, sources: impl IntoIterator<Item = Arc<dyn FeedSource>>) -> Self {
30        self.pulled.extend(sources);
31        self
32    }
33
34    /// Kept off `Worker`'s own methods, which only run what is registered.
35    #[must_use]
36    pub fn attach(self, mut worker: Worker) -> Worker {
37        self.register(&mut worker.registry);
38        worker.feeds = self;
39        worker
40    }
41
42    fn register(&self, registry: &mut Registry) {
43        for event in &self.published {
44            // Named after its event: one feed job per event, and no event is also a listener's name.
45            registry.add_taking_pruned(Entry {
46                name: event,
47                trigger: Trigger::On(event),
48                run: crate::feed::append(event),
49            });
50        }
51        if !self.published.is_empty() && registry.job(PRUNE).is_none() {
52            registry.add(Entry {
53                name: PRUNE,
54                trigger: Trigger::Every(Duration::from_hours(1).into()),
55                run: crate::feed::prune(),
56            });
57        }
58    }
59
60    pub(super) fn refuse_unkept(&self, db: &Db) -> Result<()> {
61        if self.published.is_empty() && self.pulled.is_empty() {
62            return Ok(());
63        }
64        crate::feed::feed_sql(db.driver().dialect()).map(drop)
65    }
66
67    pub(super) async fn pull_all(&self, db: &Db) -> Result<()> {
68        for source in &self.pulled {
69            while crate::feed::pull_once(db, source.as_ref()).await? {}
70        }
71        Ok(())
72    }
73
74    /// Each pull finishes the page it is on once `shutdown` completes.
75    pub(super) fn start(
76        &self,
77        db: &Db,
78        shutdown: impl Future<Output = ()> + Send,
79    ) -> (impl Future<Output = ()> + Send, Pulling) {
80        let (stop, stopped) = watch::channel(false);
81        let pulls = self
82            .pulled
83            .iter()
84            .map(|source| {
85                let pull = crate::feed::pull(db.clone(), Arc::clone(source), stopped.clone());
86                tokio::spawn(pull)
87            })
88            .collect();
89        let shutdown = async move {
90            shutdown.await;
91            let _ = stop.send(true);
92        };
93        (shutdown, Pulling(pulls))
94    }
95}
96
97pub(super) struct Pulling(Vec<JoinHandle<()>>);
98
99impl Pulling {
100    pub(super) async fn finish(self) -> Result<()> {
101        for pull in self.0 {
102            pull.await
103                .map_err(|error| Error::internal("pulling another app's events", error))?;
104        }
105        Ok(())
106    }
107}
sofie_db/worker/feeds.rs

sofie_db/worker/
feeds.rs
