refactor: Reorganise file structure

Signed-off-by: quexeky <git@quexeky.dev>

src-tauri/src/download_manager/util/download_thread_control_flag.rs

@@ -0,0 +1,49 @@
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+
+#[derive(PartialEq, Eq, PartialOrd, Ord)]
+pub enum DownloadThreadControlFlag {
+    Stop,
+    Go,
+}
+/// Go => true
+/// Stop => false
+impl From<DownloadThreadControlFlag> for bool {
+    fn from(value: DownloadThreadControlFlag) -> Self {
+        match value {
+            DownloadThreadControlFlag::Go => true,
+            DownloadThreadControlFlag::Stop => false,
+        }
+    }
+}
+/// true => Go
+/// false => Stop
+impl From<bool> for DownloadThreadControlFlag {
+    fn from(value: bool) -> Self {
+        match value {
+            true => DownloadThreadControlFlag::Go,
+            false => DownloadThreadControlFlag::Stop,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct DownloadThreadControl {
+    inner: Arc<AtomicBool>,
+}
+
+impl DownloadThreadControl {
+    pub fn new(flag: DownloadThreadControlFlag) -> Self {
+        Self {
+            inner: Arc::new(AtomicBool::new(flag.into())),
+        }
+    }
+    pub fn get(&self) -> DownloadThreadControlFlag {
+        self.inner.load(Ordering::Relaxed).into()
+    }
+    pub fn set(&self, flag: DownloadThreadControlFlag) {
+        self.inner.store(flag.into(), Ordering::Relaxed);
+    }
+}

src-tauri/src/download_manager/util/mod.rs

@@ -0,0 +1,4 @@
+pub mod progress_object;
+pub mod queue;
+pub mod rolling_progress_updates;
+pub mod download_thread_control_flag;

src-tauri/src/download_manager/util/progress_object.rs

@@ -0,0 +1,157 @@
+use std::{
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        mpsc::Sender,
+        Arc, Mutex,
+    },
+    time::{Duration, Instant},
+};
+
+use atomic_instant_full::AtomicInstant;
+use throttle_my_fn::throttle;
+
+use crate::download_manager::download_manager::DownloadManagerSignal;
+
+use super::{
+    rolling_progress_updates::RollingProgressWindow,
+};
+
+#[derive(Clone)]
+pub struct ProgressObject {
+    max: Arc<Mutex<usize>>,
+    progress_instances: Arc<Mutex<Vec<Arc<AtomicUsize>>>>,
+    start: Arc<Mutex<Instant>>,
+    sender: Sender<DownloadManagerSignal>,
+    //last_update: Arc<RwLock<Instant>>,
+    last_update_time: Arc<AtomicInstant>,
+    bytes_last_update: Arc<AtomicUsize>,
+    rolling: RollingProgressWindow<250>,
+}
+
+pub struct ProgressHandle {
+    progress: Arc<AtomicUsize>,
+    progress_object: Arc<ProgressObject>,
+}
+
+impl ProgressHandle {
+    pub fn new(progress: Arc<AtomicUsize>, progress_object: Arc<ProgressObject>) -> Self {
+        Self {
+            progress,
+            progress_object,
+        }
+    }
+    pub fn set(&self, amount: usize) {
+        self.progress.store(amount, Ordering::Relaxed);
+    }
+    pub fn add(&self, amount: usize) {
+        self.progress
+            .fetch_add(amount, std::sync::atomic::Ordering::Relaxed);
+        calculate_update(&self.progress_object);
+    }
+    pub fn skip(&self, amount: usize) {
+        self.progress
+            .fetch_add(amount, std::sync::atomic::Ordering::Relaxed);
+        // Offset the bytes at last offset by this amount
+        self.progress_object
+            .bytes_last_update
+            .fetch_add(amount, Ordering::Relaxed);
+        // Dont' fire update
+    }
+}
+
+impl ProgressObject {
+    pub fn new(max: usize, length: usize, sender: Sender<DownloadManagerSignal>) -> Self {
+        let arr = Mutex::new((0..length).map(|_| Arc::new(AtomicUsize::new(0))).collect());
+        // TODO: consolidate this calculation with the set_max function below
+        Self {
+            max: Arc::new(Mutex::new(max)),
+            progress_instances: Arc::new(arr),
+            start: Arc::new(Mutex::new(Instant::now())),
+            sender,
+
+            last_update_time: Arc::new(AtomicInstant::now()),
+            bytes_last_update: Arc::new(AtomicUsize::new(0)),
+            rolling: RollingProgressWindow::new(),
+        }
+    }
+
+    pub fn set_time_now(&self) {
+        *self.start.lock().unwrap() = Instant::now();
+    }
+    pub fn sum(&self) -> usize {
+        self.progress_instances
+            .lock()
+            .unwrap()
+            .iter()
+            .map(|instance| instance.load(Ordering::Relaxed))
+            .sum()
+    }
+    pub fn get_max(&self) -> usize {
+        *self.max.lock().unwrap()
+    }
+    pub fn set_max(&self, new_max: usize) {
+        *self.max.lock().unwrap() = new_max;
+    }
+    pub fn set_size(&self, length: usize) {
+        *self.progress_instances.lock().unwrap() =
+            (0..length).map(|_| Arc::new(AtomicUsize::new(0))).collect();
+    }
+    pub fn get_progress(&self) -> f64 {
+        self.sum() as f64 / self.get_max() as f64
+    }
+    pub fn get(&self, index: usize) -> Arc<AtomicUsize> {
+        self.progress_instances.lock().unwrap()[index].clone()
+    }
+    fn update_window(&self, kilobytes_per_second: usize) {
+        self.rolling.update(kilobytes_per_second);
+    }
+}
+
+#[throttle(1, Duration::from_millis(20))]
+pub fn calculate_update(progress: &ProgressObject) {
+    let last_update_time = progress
+        .last_update_time
+        .swap(Instant::now(), Ordering::SeqCst);
+    let time_since_last_update = Instant::now().duration_since(last_update_time).as_millis();
+
+    let current_bytes_downloaded = progress.sum();
+    let max = progress.get_max();
+    let bytes_at_last_update = progress
+        .bytes_last_update
+        .swap(current_bytes_downloaded, Ordering::Relaxed);
+
+    let bytes_since_last_update = current_bytes_downloaded - bytes_at_last_update;
+
+    let kilobytes_per_second = bytes_since_last_update / (time_since_last_update as usize).max(1);
+
+    let bytes_remaining = max - current_bytes_downloaded; // bytes
+
+    progress.update_window(kilobytes_per_second);
+    push_update(progress, bytes_remaining);
+}
+
+#[throttle(1, Duration::from_millis(500))]
+pub fn push_update(progress: &ProgressObject, bytes_remaining: usize) {
+    let average_speed = progress.rolling.get_average();
+    let time_remaining = (bytes_remaining / 1000) / average_speed.max(1);
+
+    update_ui(progress, average_speed, time_remaining);
+    update_queue(progress);
+}
+
+fn update_ui(progress_object: &ProgressObject, kilobytes_per_second: usize, time_remaining: usize) {
+    progress_object
+        .sender
+        .send(DownloadManagerSignal::UpdateUIStats(
+            kilobytes_per_second,
+            time_remaining,
+        ))
+        .unwrap();
+}
+
+fn update_queue(progress: &ProgressObject) {
+    progress
+        .sender
+        .send(DownloadManagerSignal::UpdateUIQueue)
+        .unwrap();
+}

src-tauri/src/download_manager/util/queue.rs

@@ -0,0 +1,80 @@
+use std::{
+    collections::VecDeque,
+    sync::{Arc, Mutex, MutexGuard},
+};
+
+use crate::database::models::data::DownloadableMetadata;
+
+#[derive(Clone)]
+pub struct Queue {
+    inner: Arc<Mutex<VecDeque<DownloadableMetadata>>>,
+}
+
+#[allow(dead_code)]
+impl Default for Queue {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Queue {
+    pub fn new() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(VecDeque::new())),
+        }
+    }
+    pub fn read(&self) -> VecDeque<DownloadableMetadata> {
+        self.inner.lock().unwrap().clone()
+    }
+    pub fn edit(&self) -> MutexGuard<'_, VecDeque<DownloadableMetadata>> {
+        self.inner.lock().unwrap()
+    }
+    pub fn pop_front(&self) -> Option<DownloadableMetadata> {
+        self.edit().pop_front()
+    }
+    pub fn is_empty(&self) -> bool {
+        self.inner.lock().unwrap().len() == 0
+    }
+    pub fn exists(&self, meta: DownloadableMetadata) -> bool {
+        self.read().contains(&meta)
+    }
+    /// Either inserts `interface` at the specified index, or appends to
+    /// the back of the deque if index is greater than the length of the deque
+    pub fn insert(&self, interface: DownloadableMetadata, index: usize) {
+        if self.read().len() > index {
+            self.append(interface);
+        } else {
+            self.edit().insert(index, interface);
+        }
+    }
+    pub fn append(&self, interface: DownloadableMetadata) {
+        self.edit().push_back(interface);
+    }
+    pub fn pop_front_if_equal(&self, meta: &DownloadableMetadata) -> Option<DownloadableMetadata> {
+        let mut queue = self.edit();
+        let front = queue.front()?;
+        if front == meta {
+            return queue.pop_front();
+        }
+        None
+    }
+    pub fn get_by_meta(&self, meta: &DownloadableMetadata) -> Option<usize> {
+        self.read().iter().position(|data| data == meta)
+    }
+    pub fn move_to_index_by_meta(
+        &self,
+        meta: &DownloadableMetadata,
+        new_index: usize,
+    ) -> Result<(), ()> {
+        let index = match self.get_by_meta(meta) {
+            Some(index) => index,
+            None => return Err(()),
+        };
+        let existing = match self.edit().remove(index) {
+            Some(existing) => existing,
+            None => return Err(()),
+        };
+        self.edit().insert(new_index, existing);
+        Ok(())
+    }
+}

src-tauri/src/download_manager/util/rolling_progress_updates.rs

@@ -0,0 +1,33 @@
+use std::sync::{
+    atomic::{AtomicUsize, Ordering},
+    Arc,
+};
+
+#[derive(Clone)]
+pub struct RollingProgressWindow<const S: usize> {
+    window: Arc<[AtomicUsize; S]>,
+    current: Arc<AtomicUsize>,
+}
+impl<const S: usize> RollingProgressWindow<S> {
+    pub fn new() -> Self {
+        Self {
+            window: Arc::new([(); S].map(|_| AtomicUsize::new(0))),
+            current: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+    pub fn update(&self, kilobytes_per_second: usize) {
+        let index = self.current.fetch_add(1, Ordering::SeqCst);
+        let current = &self.window[index % S];
+        current.store(kilobytes_per_second, Ordering::SeqCst);
+    }
+    pub fn get_average(&self) -> usize {
+        let current = self.current.load(Ordering::SeqCst);
+        self.window
+            .iter()
+            .enumerate()
+            .filter(|(i, _)| i < &current)
+            .map(|(_, x)| x.load(Ordering::Relaxed))
+            .sum::<usize>()
+            / S
+    }
+}