gix_worktree_stream/from_tree/
mod.rs

1use std::io::Write;
2
3use gix_object::{bstr::BStr, FindExt};
4
5use crate::{entry, entry::Error, protocol, AdditionalEntry, SharedErrorSlot, Stream};
6
7/// Use `objects` to traverse `tree` and fetch the contained blobs to return as [`Stream`], which makes them queryable
8/// on demand with support for streaming each entry.
9///
10/// `pipeline` is used to convert blobs to their worktree representation, and `attributes` is used to read
11/// the `export-ignore` attribute. If set on a directory or blob, it won't be added to the archive.
12///
13/// ### Types of entries in stream
14///
15/// We only return blobs (with or without executable), which may be symlinks in which case their content will
16/// be target of the symlink.
17/// Directories are never returned, but maybe added by the caller via [Stream::add_entry()].
18///
19/// ### Progress and interruptions
20///
21/// For per-file progress, integrate progress handling into the calls of [`Stream::next_entry()`] as that
22/// correlates blobs.
23/// Additional interrupt handling can be wrapped around the `Read` implementation of each [`Entry`][crate::Entry].
24/// For progress on bytes-written, integrate progress reporting when consuming the stream.
25/// Further it's possible to drop the returned [`Stream`] to halt all operation.
26///
27/// ### Threaded Operation
28///
29/// This function spawns a thread that will access the tree data in the background, synchronized through
30/// `Stream` so that it will not be faster than the consumer, with at most one file in flight at any time.
31///
32/// ### Limitations
33///
34/// * `export-subst` is not support, as it requires the entire formatting engine of `git log`.
35pub fn from_tree<Find, E>(
36    tree: gix_hash::ObjectId,
37    objects: Find,
38    pipeline: gix_filter::Pipeline,
39    attributes: impl FnMut(&BStr, gix_object::tree::EntryMode, &mut gix_attributes::search::Outcome) -> Result<(), E>
40        + Send
41        + 'static,
42) -> Stream
43where
44    Find: gix_object::Find + Clone + Send + 'static,
45    E: std::error::Error + Send + Sync + 'static,
46{
47    let (stream, mut write, additional_entries) = Stream::new();
48    std::thread::spawn({
49        let slot = stream.err.clone();
50        move || {
51            if let Err(err) = run(
52                tree,
53                objects,
54                pipeline,
55                attributes,
56                &mut write,
57                slot.clone(),
58                additional_entries,
59            ) {
60                {
61                    let mut slot = slot.lock();
62                    if slot.is_none() {
63                        *slot = Some(err);
64                    } else {
65                        drop(slot);
66                        write.channel.send(Err(std::io::Error::other(err))).ok();
67                    }
68                }
69            }
70        }
71    });
72    stream
73}
74
75fn run<Find, E>(
76    tree: gix_hash::ObjectId,
77    objects: Find,
78    mut pipeline: gix_filter::Pipeline,
79    mut attributes: impl FnMut(&BStr, gix_object::tree::EntryMode, &mut gix_attributes::search::Outcome) -> Result<(), E>
80        + Send
81        + 'static,
82    out: &mut gix_features::io::pipe::Writer,
83    err: SharedErrorSlot,
84    additional_entries: std::sync::mpsc::Receiver<AdditionalEntry>,
85) -> Result<(), Error>
86where
87    Find: gix_object::Find + Clone,
88    E: std::error::Error + Send + Sync + 'static,
89{
90    let mut buf = Vec::new();
91    let tree_iter = objects.find_tree_iter(tree.as_ref(), &mut buf)?;
92    if pipeline.driver_context_mut().treeish.is_none() {
93        pipeline.driver_context_mut().treeish = Some(tree);
94    }
95
96    let mut attrs = gix_attributes::search::Outcome::default();
97    attrs.initialize_with_selection(&Default::default(), Some("export-ignore"));
98    let mut dlg = traverse::Delegate {
99        out,
100        err,
101        pipeline,
102        attrs,
103        objects: objects.clone(),
104        fetch_attributes: move |a: &BStr, b: gix_object::tree::EntryMode, c: &mut gix_attributes::search::Outcome| {
105            attributes(a, b, c).map_err(|err| Error::Attributes {
106                source: Box::new(err),
107                path: a.to_owned(),
108            })
109        },
110        path_deque: Default::default(),
111        path: Default::default(),
112        buf: Vec::with_capacity(1024),
113    };
114    gix_traverse::tree::breadthfirst(
115        tree_iter,
116        gix_traverse::tree::breadthfirst::State::default(),
117        &objects,
118        &mut dlg,
119    )?;
120
121    for entry in additional_entries {
122        protocol::write_entry_header_and_path(
123            entry.relative_path.as_ref(),
124            &entry.id,
125            entry.mode,
126            entry.source.len(),
127            out,
128        )?;
129        // pipe writer always writes all in one go.
130        #[allow(clippy::unused_io_amount)]
131        match entry.source {
132            entry::Source::Memory(buf) => out.write(&buf).map(|_| ()),
133            entry::Source::Null => out.write(&[]).map(|_| ()),
134            entry::Source::Path(path) => {
135                let file = std::fs::File::open(path)?;
136                protocol::write_stream(&mut buf, file, out)
137            }
138        }?;
139    }
140    Ok(())
141}
142
143mod traverse;
gix_worktree_stream/from_tree/mod.rs

gix_worktree_stream/from_tree/
mod.rs
