nifty_download/
lib.rs

1//! `download(…)`, shared by the interpreter and compiled apps: each download runs on its own
2//! thread and sends reports back over a channel, waking the UI thread, which writes them
3//! wherever the app's call bound them.
4
5mod connection;
6mod follow;
7mod http;
8mod id;
9mod name;
10mod part;
11mod report;
12mod retry;
13mod size;
14mod speed;
15#[cfg(feature = "testing")]
16pub mod testing;
17mod worker;
18
19use std::cell::RefCell;
20use std::collections::HashMap;
21#[cfg(feature = "testing")]
22use std::collections::VecDeque;
23use std::sync::Arc;
24use std::sync::mpsc::{self, Receiver, Sender};
25use std::thread::{self, JoinHandle};
26
27use nifty_runtime::{DownloadCommand, Waker};
28
29pub use follow::{StatePlace, StateReader, follow_id};
30pub use id::DownloadId;
31pub use report::{Progress, Report, Written};
32pub use size::format_bytes;
33use worker::{Abandoned, Event, Job};
34
35/// A download an action asked for, started once the action has finished. `targets` says
36/// where its reports go, in whatever terms the app uses.
37#[derive(Debug)]
38pub struct Request<T> {
39    pub id: DownloadId,
40    pub url: String,
41    pub folder: String,
42    /// The name to save the file as, made safe first; `""` names it after the address.
43    pub name: String,
44    /// How many connections the call asked for; any number below one asks for one.
45    pub connections: i64,
46    /// Headers to send with each request, each written `Name: value`.
47    pub headers: Vec<String>,
48    /// How many times in a row a failed connection is tried again, when the failure may
49    /// pass; any number below zero asks for none.
50    pub retries: i64,
51    pub targets: T,
52}
53
54/// The downloads an app has running, and the channel their reports arrive on.
55pub struct Downloads<T> {
56    waker: Arc<dyn Waker>,
57    sender: Sender<(DownloadId, Report)>,
58    receiver: Receiver<(DownloadId, Report)>,
59    running: RefCell<HashMap<DownloadId, Running<T>>>,
60    /// Reports a test waited for, handed out before any others.
61    #[cfg(feature = "testing")]
62    waited: RefCell<VecDeque<(DownloadId, Report)>>,
63}
64
65type Worker = JoinHandle<Result<(), Abandoned>>;
66
67struct Running<T> {
68    targets: T,
69    /// Where the app's commands for it go.
70    events: Sender<Event>,
71    /// Whether the app still holds it. One whose item is gone is cancelled, and its last
72    /// reports go nowhere.
73    held: bool,
74    /// Joined once its last report arrives; `None` when a test is already waiting on it.
75    worker: Option<Worker>,
76}
77
78impl<T> Downloads<T> {
79    /// `waker` asks the UI thread to call [`Downloads::receive`].
80    pub fn new(waker: Arc<dyn Waker>) -> Self {
81        let (sender, receiver) = mpsc::channel();
82        Self {
83            waker,
84            sender,
85            receiver,
86            running: RefCell::new(HashMap::new()),
87            #[cfg(feature = "testing")]
88            waited: RefCell::new(VecDeque::new()),
89        }
90    }
91
92    pub fn start(&self, request: Request<T>) {
93        let Request {
94            id,
95            url,
96            folder,
97            name,
98            connections,
99            headers,
100            retries,
101            targets,
102        } = request;
103        let job = Job {
104            url,
105            folder,
106            name: (!name.is_empty()).then_some(name),
107            connections: part::connection_count(connections),
108            headers,
109            retries: retry::retry_count(retries),
110        };
111        let (events, inbox) = mpsc::channel();
112        let connect = events.clone();
113        let sender = self.sender.clone();
114        let waker = Arc::clone(&self.waker);
115        let worker = thread::spawn(move || {
116            worker::run(job, &inbox, connect, &|report| {
117                sender.send((id, report)).map_err(|_| Abandoned)?;
118                waker.wake();
119                Ok(())
120            })
121        });
122        self.running.borrow_mut().insert(
123            id,
124            Running {
125                targets,
126                events,
127                held: true,
128                worker: Some(worker),
129            },
130        );
131    }
132
133    /// Asks the download whose id `id` shows to do `command`. An id of no running download,
134    /// such as one that has just finished, asks nothing.
135    pub fn command(&self, id: &str, command: DownloadCommand) {
136        let running = self.running.borrow();
137        let download = DownloadId::parse(id).and_then(|id| running.get(&id));
138        if let Some(download) = download.filter(|download| download.held) {
139            download.send(Event::Command(command));
140        }
141    }
142
143    /// Hands each report that has arrived to `apply` with where it goes, forgetting
144    /// downloads that have finished. `follow` first moves a download's targets to where its
145    /// item is now, or says the app no longer holds it, which cancels it. Stops at the
146    /// first error, leaving later reports queued.
147    ///
148    /// # Errors
149    /// The first error `apply` returns.
150    ///
151    /// # Panics
152    /// When a report arrives from a download this doesn't know, which can't happen: each
153    /// is forgotten only after its last report.
154    pub fn receive<E>(
155        &self,
156        mut follow: impl FnMut(&mut T) -> bool,
157        mut apply: impl FnMut(&T, &Report) -> Result<(), E>,
158    ) -> Result<(), E> {
159        while let Some((id, report)) = self.next_report() {
160            let mut running = self.running.borrow_mut();
161            let download = running
162                .get_mut(&id)
163                .expect("reports only come from downloads still running");
164            download.keep_if_held(&mut follow);
165            let applied = if download.held {
166                apply(&download.targets, &report)
167            } else {
168                Ok(())
169            };
170            if report.is_final()
171                && let Some(finished) = running.remove(&id)
172            {
173                finished.join();
174            }
175            applied?;
176        }
177        Ok(())
178    }
179
180    /// Cancels each download whose item `follow` no longer finds, after the app's code
181    /// changed state, and keeps the rest's targets where their items are.
182    pub fn cancel_unheld(&self, mut follow: impl FnMut(&mut T) -> bool) {
183        for download in self.running.borrow_mut().values_mut() {
184            download.keep_if_held(&mut follow);
185        }
186    }
187
188    #[cfg(not(feature = "testing"))]
189    fn next_report(&self) -> Option<(DownloadId, Report)> {
190        self.receiver.try_recv().ok()
191    }
192
193    #[cfg(feature = "testing")]
194    fn next_report(&self) -> Option<(DownloadId, Report)> {
195        let waited = self.waited.borrow_mut().pop_front();
196        waited.or_else(|| self.receiver.try_recv().ok())
197    }
198
199    /// Waits until a report arrives, keeps it for [`Downloads::receive`], and wakes the
200    /// UI thread to receive it, in case the download hasn't yet.
201    ///
202    /// # Panics
203    /// Never: this holds a sender, so receiving can't disconnect.
204    #[cfg(feature = "testing")]
205    pub fn wait_for_report(&self) {
206        let report = self
207            .receiver
208            .recv()
209            .expect("this holds a sender, so receiving never disconnects");
210        self.waited.borrow_mut().push_back(report);
211        self.waker.wake();
212    }
213
214    /// Waits until every running download's worker has sent its last report.
215    #[cfg(feature = "testing")]
216    pub fn wait(&self) {
217        let workers: Vec<_> = self
218            .running
219            .borrow_mut()
220            .values_mut()
221            .filter_map(|download| download.worker.take())
222            .collect();
223        for worker in workers {
224            join(worker);
225        }
226    }
227}
228
229impl<T> Running<T> {
230    /// Cancels this download if `follow` finds the app no longer holds it.
231    fn keep_if_held(&mut self, follow: &mut impl FnMut(&mut T) -> bool) {
232        if self.held && !follow(&mut self.targets) {
233            self.held = false;
234            self.send(Event::Command(DownloadCommand::Cancel));
235        }
236    }
237
238    fn send(&self, event: Event) {
239        // This only fails once the worker has sent its last report and stopped listening,
240        // when there is nothing left to ask of it.
241        let _ = self.events.send(event);
242    }
243
244    /// Waits for the worker, which exits right after sending its last report.
245    fn join(self) {
246        if let Some(worker) = self.worker {
247            join(worker);
248        }
249    }
250}
251
252fn join(worker: Worker) {
253    let finished = worker.join().expect("download workers don't panic");
254    assert!(
255        finished.is_ok(),
256        "a worker is only abandoned once its app is gone"
257    );
258}
259
260/// The app is going, as when `nf dev` replaces its program: each download stops and
261/// removes its partial file before this returns. A download's thread never waits on the
262/// network, only on its events, so this is quick.
263impl<T> Drop for Downloads<T> {
264    fn drop(&mut self) {
265        let running = self.running.get_mut();
266        for download in running.values() {
267            download.send(Event::Abandoned);
268        }
269        for worker in running
270            .values_mut()
271            .filter_map(|download| download.worker.take())
272        {
273            // Abandoned is how a worker ends once its app is gone, unless it had already
274            // finished.
275            let _: Result<(), Abandoned> = worker.join().expect("download workers don't panic");
276        }
277    }
278}
279
280impl<T> std::fmt::Debug for Downloads<T> {
281    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
282        f.debug_struct("Downloads")
283            .field("running", &self.running.borrow().len())
284            .finish_non_exhaustive()
285    }
286}
nifty_download/lib.rs

nifty_download/
lib.rs
