use crate::utils::ForceSendExt;
use crate::{DownloadState, Event, StateError};
use crate::{PartialConfig, TerminationReason, Tx};
use fast_down::UrlInfo;
use std::path::Path;
use tokio::fs::{self, OpenOptions};
use tokio_util::sync::CancellationToken;
use url::Url;
mod overwrite;
mod pipeline;
mod plan;
mod progress_reporter;
pub use plan::*;
fn open_existing() -> OpenOptions {
let mut o = OpenOptions::new();
o.read(true).write(true).truncate(false).create(false);
o
}
fn open_create() -> OpenOptions {
let mut o = OpenOptions::new();
o.read(true).write(true).truncate(false).create(true);
o
}
fn open_create_new() -> OpenOptions {
let mut o = OpenOptions::new();
o.read(true).write(true).truncate(false).create_new(true);
o
}
#[allow(clippy::result_large_err)]
async fn try_load_resume_state(
url: &Url,
cfg_path: &Path,
tmp_path: &Path,
info: &UrlInfo,
partial_config: &PartialConfig,
) -> Result<Option<DownloadState>, StateError> {
let fd_exists = fs::try_exists(cfg_path).await.unwrap_or(false);
let tmp_exists = fs::try_exists(tmp_path).await.unwrap_or(false);
if !fd_exists || !tmp_exists {
return Ok(None);
}
let state = DownloadState::load(cfg_path).await?;
state.validate(info)?;
state.merge_config(partial_config);
if state.part_shortfall(tmp_path).await.is_some() {
return Ok(None);
}
state.refresh_identity(url, info);
Ok(Some(state))
}
pub fn download(url: Url, partial_config: PartialConfig, tx: Tx, token: CancellationToken) {
tokio::spawn(
async move {
let token2 = token.clone();
let planned =
Box::pin(token.run_until_cancelled(plan(url, partial_config, tx.clone(), token2)))
.await;
Box::pin(drive(planned, &tx)).await;
}
.force_send(),
);
}
pub fn resume(
tmp_path: impl AsRef<Path>,
url: Option<Url>,
partial_config: PartialConfig,
tx: Tx,
token: CancellationToken,
) {
let tmp_path = tmp_path.as_ref().to_path_buf();
tokio::spawn(
async move {
let token2 = token.clone();
let planned = Box::pin(token.run_until_cancelled(plan_resume(
tmp_path,
url,
partial_config,
tx.clone(),
token2,
)))
.await;
Box::pin(drive(planned, &tx)).await;
}
.force_send(),
);
}
pub fn download_from_fd(
fd_path: impl AsRef<Path>,
url: Option<Url>,
partial_config: PartialConfig,
tx: Tx,
token: CancellationToken,
) {
let fd_path = fd_path.as_ref().to_path_buf();
tokio::spawn(
async move {
let token2 = token.clone();
let planned = Box::pin(token.run_until_cancelled(plan_from_fd(
fd_path,
url,
partial_config,
tx.clone(),
token2,
)))
.await;
Box::pin(drive(planned, &tx)).await;
}
.force_send(),
);
}
async fn drive(planned: Option<Result<DownloadPlan, PlanError>>, tx: &Tx) {
match planned {
None => {
let _ = tx.send(Event::Terminated(TerminationReason::Cancelled));
}
Some(Err(e)) => {
e.emit(tx);
let _ = tx.send(Event::Terminated(TerminationReason::Failed));
}
Some(Ok(prepared)) => Box::pin(prepared.start()).await,
}
}