nifty_download/
connection.rs

1//! One connection to the server, on its own thread: it asks for the file and writes the
2//! body into the file the download gives it, telling the download each step.
3//!
4//! A connection blocked waiting for the server can't be interrupted, so the download
5//! never waits for one: it closes the connection's [`Outlet`], and the connection stops at
6//! its next write without touching the file again.
7
8use std::fs::File;
9use std::io::{self, Read, Write};
10use std::sync::mpsc::Sender;
11use std::sync::{Arc, Condvar, Mutex, MutexGuard};
12use std::thread;
13
14use ureq::{Agent, Timeout};
15
16use crate::http::{self, ByteRange, Headers};
17use crate::worker::Event;
18
19const CHUNK_SIZE: usize = 64 * 1024;
20
21/// Which of a download's connections an event comes from, counted from 0, so events from
22/// one it has stopped can be told apart from the one it runs now.
23pub(crate) type ConnectionId = usize;
24
25/// Where a connection writes: nowhere yet, the download's file, or nowhere ever again.
26#[derive(Debug, Default)]
27pub(crate) struct Outlet {
28    inner: Mutex<Inner>,
29    changed: Condvar,
30}
31
32#[derive(Debug, Default)]
33struct Inner {
34    slot: Slot,
35    /// Bytes written through it so far.
36    written: u64,
37    /// How many more bytes it takes before it is full, when its connection fetches a part
38    /// that ends before the file does.
39    room: Option<u64>,
40}
41
42#[derive(Debug, Default)]
43enum Slot {
44    #[default]
45    Waiting,
46    Open(File),
47    Closed,
48}
49
50/// Whether a connection can keep writing after a write.
51#[derive(Debug, PartialEq, Eq)]
52enum Flow {
53    More,
54    /// It wrote the last byte of its part.
55    Full,
56    /// The download closed it, or stopped listening.
57    Stopped,
58}
59
60/// What a closed outlet leaves: the file it was writing, if it had opened, and how many
61/// bytes went through it.
62#[derive(Debug)]
63pub(crate) struct Closed {
64    pub file: Option<File>,
65    pub written: u64,
66}
67
68impl Outlet {
69    /// Lets the connection write into `file`, from its current position on, and at most
70    /// `room` bytes when that is known.
71    pub(crate) fn open(&self, file: File, room: Option<u64>) {
72        let mut inner = self.lock();
73        inner.slot = Slot::Open(file);
74        inner.room = room;
75        drop(inner);
76        self.changed.notify_all();
77    }
78
79    /// Stops the connection writing. Once this returns, the connection won't write again.
80    pub(crate) fn close(&self) -> Closed {
81        let mut inner = self.lock();
82        let slot = std::mem::replace(&mut inner.slot, Slot::Closed);
83        let written = inner.written;
84        drop(inner);
85        self.changed.notify_all();
86        let file = match slot {
87            Slot::Open(file) => Some(file),
88            Slot::Waiting | Slot::Closed => None,
89        };
90        Closed { file, written }
91    }
92
93    /// Writes `bytes`, or as many as there is room for, once the outlet opens, and tells
94    /// `events`, both before it can close.
95    fn write(&self, bytes: &[u8], id: ConnectionId, events: &Sender<Event>) -> io::Result<Flow> {
96        let mut inner = self.lock();
97        while matches!(inner.slot, Slot::Waiting) {
98            inner = self
99                .changed
100                .wait(inner)
101                .expect("nothing panics while holding an outlet");
102        }
103        let room = inner.room;
104        let Slot::Open(file) = &mut inner.slot else {
105            return Ok(Flow::Stopped);
106        };
107        let len = room.map_or(bytes.len(), |room| {
108            bytes.len().min(usize::try_from(room).unwrap_or(usize::MAX))
109        });
110        let fits = &bytes[..len];
111        file.write_all(fits)?;
112        let count = fits.len() as u64;
113        inner.written += count;
114        let room = room.map(|room| room - count);
115        inner.room = room;
116        if events.send(Event::Wrote(id, count)).is_err() {
117            Ok(Flow::Stopped)
118        } else if room == Some(0) {
119            Ok(Flow::Full)
120        } else {
121            Ok(Flow::More)
122        }
123    }
124
125    fn lock(&self) -> MutexGuard<'_, Inner> {
126        self.inner
127            .lock()
128            .expect("nothing panics while holding an outlet")
129    }
130}
131
132/// What a connection asks for: `url`, or just `range` of it, with the app's `headers`.
133#[derive(Debug)]
134pub(crate) struct Dial {
135    pub id: ConnectionId,
136    pub url: String,
137    pub range: Option<ByteRange>,
138    pub headers: Headers,
139}
140
141/// Starts a connection asking `agent` for what `dial` says, which writes into `outlet`
142/// and tells `events` how it goes.
143pub(crate) fn spawn(agent: Agent, dial: Dial, outlet: Arc<Outlet>, events: Sender<Event>) {
144    thread::spawn(move || {
145        let id = dial.id;
146        let ended = match http::get(&agent, &dial.url, dial.range, &dial.headers) {
147            Ok((answer, body)) => {
148                if events.send(Event::Answered(id, answer)).is_err() {
149                    return;
150                }
151                copy(body.into_reader(), id, &outlet, &events)
152            }
153            Err(error) => Err(error),
154        };
155        // Once the download has moved on, nobody needs to hear how this one ended.
156        if let Some(ended) = ended.transpose() {
157            let _ = events.send(Event::Ended(id, ended));
158        }
159    });
160}
161
162/// Copies `body` through `outlet`: `Some(())` at the end of the body or once the outlet is
163/// full, `None` when the download stopped listening first.
164fn copy(
165    mut body: impl Read,
166    id: ConnectionId,
167    outlet: &Outlet,
168    events: &Sender<Event>,
169) -> Result<Option<()>, ConnectionError> {
170    let mut buffer = vec![0; CHUNK_SIZE];
171    loop {
172        let read = body.read(&mut buffer).map_err(ConnectionError::Read)?;
173        if read == 0 {
174            return Ok(Some(()));
175        }
176        match outlet
177            .write(&buffer[..read], id, events)
178            .map_err(ConnectionError::Write)?
179        {
180            Flow::More => {}
181            Flow::Full => return Ok(Some(())),
182            Flow::Stopped => return Ok(None),
183        }
184    }
185}
186
187/// Why a connection stopped early.
188#[derive(Debug)]
189pub(crate) enum ConnectionError {
190    Http(ureq::Error),
191    /// Reading the body from the server failed.
192    Read(io::Error),
193    /// Writing the file failed.
194    Write(io::Error),
195    /// The server sent part of the file without saying which part.
196    UnclearPart,
197}
198
199impl ConnectionError {
200    /// Whether trying again may work: the body broke off, connecting timed out, or the
201    /// server had a problem of its own, a `5xx` status.
202    pub(crate) fn is_transient(&self) -> bool {
203        match self {
204            Self::Read(_)
205            | Self::Http(
206                ureq::Error::Timeout(Timeout::Connect) | ureq::Error::StatusCode(500..=599),
207            ) => true,
208            Self::Http(_) | Self::Write(_) | Self::UnclearPart => false,
209        }
210    }
211}
212
213impl From<ureq::Error> for ConnectionError {
214    fn from(error: ureq::Error) -> Self {
215        Self::Http(error)
216    }
217}
218
219impl std::fmt::Display for ConnectionError {
220    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
221        match self {
222            Self::Http(error) => write!(f, "{error}"),
223            Self::Read(error) | Self::Write(error) => write!(f, "{error}"),
224            Self::UnclearPart => {
225                f.write_str("the server sent part of the file without saying which")
226            }
227        }
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use super::*;
234
235    #[test]
236    fn retries_only_what_may_pass() {
237        let io = || io::Error::other("reset");
238        assert!(ConnectionError::Read(io()).is_transient());
239        assert!(ConnectionError::Http(ureq::Error::Timeout(Timeout::Connect)).is_transient());
240        assert!(ConnectionError::Http(ureq::Error::StatusCode(503)).is_transient());
241        assert!(!ConnectionError::Http(ureq::Error::StatusCode(404)).is_transient());
242        assert!(!ConnectionError::Http(ureq::Error::Timeout(Timeout::RecvResponse)).is_transient());
243        assert!(!ConnectionError::Http(ureq::Error::HostNotFound).is_transient());
244        assert!(!ConnectionError::Write(io()).is_transient());
245        assert!(!ConnectionError::UnclearPart.is_transient());
246    }
247}
nifty_download/connection.rs

nifty_download/
connection.rs
