use std::num::NonZeroU32;
use std::ops::ControlFlow;
use std::path::Path;
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
use std::process::Command;
use std::sync::atomic::AtomicBool;
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
use super::receive_with_known;
use super::{Advertisement, FetchError, FetchLimits, KnownHistory, ReceivedFetch};
use crate::ObjectId;
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
use crate::transport::Server;
use crate::transport::TransportControl;
pub fn receive_local(
source: impl AsRef<Path>,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
limits: FetchLimits,
cancel: &AtomicBool,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, FetchError> {
receive_local_with_control(
source,
select,
limits,
TransportControl::new(cancel),
progress,
)
}
pub fn receive_local_with_control(
source: impl AsRef<Path>,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
limits: FetchLimits,
control: TransportControl<'_>,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, FetchError> {
receive_local_with_known(
source,
select,
&KnownHistory::default(),
limits,
control,
progress,
)
}
pub fn receive_local_with_known(
source: impl AsRef<Path>,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
known: &KnownHistory,
limits: FetchLimits,
control: TransportControl<'_>,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, FetchError> {
receive_local_with_known_and_progress(source, select, known, limits, control, progress, |_| {})
}
pub fn receive_local_with_known_and_progress(
source: impl AsRef<Path>,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
known: &KnownHistory,
limits: FetchLimits,
control: TransportControl<'_>,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
observe: impl FnMut(super::LocalFetchProgress),
) -> Result<ReceivedFetch, FetchError> {
receive_local_with_known_depth_and_progress(
source, select, known, None, limits, control, progress, observe,
)
}
#[expect(
clippy::too_many_arguments,
reason = "local transfer keeps explicit selection and progress policy"
)]
pub(super) fn receive_local_with_known_depth_and_progress(
source: impl AsRef<Path>,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
known: &KnownHistory,
depth: Option<NonZeroU32>,
limits: FetchLimits,
control: TransportControl<'_>,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
observe: impl FnMut(super::LocalFetchProgress),
) -> Result<ReceivedFetch, FetchError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"fetch.local",
outcome = "incomplete",
failure_class = tracing::field::Empty,
effects = tracing::field::Empty,
);
let operation = || {
control.check()?;
let source =
crate::Repository::open(source).map_err(|error| FetchError::Source(Box::new(error)))?;
super::local_native::receive_depth(
&source, select, known, depth, limits, control, progress, observe,
)
};
#[cfg(feature = "tracing")]
let result = span.in_scope(operation);
#[cfg(not(feature = "tracing"))]
let result = { operation }();
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, crate::trace::fetch);
result
}
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
fn receive_server(
command: &mut Command,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
limits: FetchLimits,
control: TransportControl<'_>,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, FetchError> {
receive_server_with_known(
command,
select,
&KnownHistory::default(),
limits,
control,
progress,
)
}
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
fn receive_server_with_known(
command: &mut Command,
select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
known: &KnownHistory,
limits: FetchLimits,
control: TransportControl<'_>,
progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, FetchError> {
#[cfg(feature = "tracing")]
let span = tracing::debug_span!(
target: "girt",
"fetch.local_server",
outcome = "incomplete",
failure_class = tracing::field::Empty,
effects = tracing::field::Empty,
);
let operation = || {
let mut child = Server::spawn(command, control)?;
let received = {
let (mut reader, mut writer) = child.streams();
receive_with_known(
&mut reader,
&mut writer,
select,
known,
limits,
control.cancel,
progress,
)?
};
let status = child.wait()?;
if !status.success() {
return Err(FetchError::Process(status));
}
Ok(received)
};
#[cfg(feature = "tracing")]
let result = span.in_scope(operation);
#[cfg(not(feature = "tracing"))]
let result = { operation }();
#[cfg(feature = "tracing")]
crate::trace::finish(&span, &result, crate::trace::fetch);
result
}
#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
mod tests;