1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
//! Batch object lookup: one request, one schedule, forward seeks only.
use std::{collections::HashSet, io};
use ltk_hash::BinHash;
use crate::{
stream::{BinStream, ObjectEntry, ObjectStream},
Error,
};
/// Lending cursor over a requested set of objects, in file order.
///
/// Created by [`BinStream::objects_batch`]. Like [`Objects`](crate::stream::Objects), each
/// yielded [`ObjectStream`] borrows the reader, so the borrow checker enforces one open object
/// at a time.
#[must_use = "cursors are lazy and read nothing until advanced"]
#[derive(Debug)]
pub struct BatchObjects<'a, R: io::Read + io::Seek> {
stream: &'a mut BinStream<R>,
/// The request, deduplicated, in the order it was given — the order [`Self::missing`]
/// reports in.
requested: Vec<BinHash>,
/// What the cursor has not found yet.
pending: HashSet<BinHash>,
missing: Vec<BinHash>,
/// The requested rows in offset order, when the table of contents was already complete.
/// `None` means the cursor is scanning the object table instead.
ordered: Option<Vec<ObjectEntry>>,
/// A table position while scanning, an index into `ordered` otherwise.
at: usize,
}
impl<'a, R: io::Read + io::Seek> BatchObjects<'a, R> {
pub(crate) fn new(
stream: &'a mut BinStream<R>,
hashes: impl IntoIterator<Item = impl Into<BinHash>>,
) -> Self {
let mut requested = Vec::new();
let mut pending = HashSet::new();
for hash in hashes {
let hash = hash.into();
if pending.insert(hash) {
requested.push(hash);
}
}
// A complete table of contents answers the whole request without reading anything, so
// the schedule is decided here: the found rows sorted by offset, the rest missing.
let (ordered, missing) = match stream.is_toc_complete() {
false => (None, Vec::new()),
true => {
let mut found = Vec::with_capacity(requested.len());
let mut missing = Vec::new();
for &hash in &requested {
match stream.toc_entry(hash) {
Some(&entry) => found.push(entry),
None => missing.push(hash),
}
}
found.sort_unstable_by_key(|entry| entry.offset);
pending.clear();
(Some(found), missing)
}
};
Self {
stream,
requested,
pending,
missing,
ordered,
at: 0,
}
}
/// Advances to the next requested object the table contains.
///
/// Absent hashes are not yielded — a miss has no file position, so it has no place in a
/// file-order sequence. [`BatchObjects::missing`] reports them once the cursor is done.
///
/// # Errors
///
/// An I/O error from the source while scanning the object table.
#[expect(
clippy::should_implement_trait,
reason = "a lending cursor: the yielded item borrows the reader, which `Iterator` cannot express"
)]
pub fn next(&mut self) -> Result<Option<ObjectStream<'_, R>>, Error> {
match self.next_entry()? {
Some(entry) => Ok(Some(ObjectStream::new(self.stream, entry))),
None => Ok(None),
}
}
/// The requested hashes the object table does not contain.
///
/// Complete once [`BatchObjects::next`] has returned `Ok(None)`; before that it only holds
/// what the cursor has already ruled out, which for a scan is nothing.
#[must_use]
pub fn missing(&self) -> &[BinHash] {
&self.missing
}
/// The next row to open, by whichever schedule this cursor is on.
fn next_entry(&mut self) -> Result<Option<ObjectEntry>, Error> {
if let Some(ordered) = &self.ordered {
let entry = ordered.get(self.at).copied();
if entry.is_some() {
self.at += 1;
}
return Ok(entry);
}
let total = self.stream.class_hashes().len();
// Nothing left to look for — an empty request, or one the scan already answered — so
// there is no reason to read any more of the table.
if self.pending.is_empty() {
self.at = total;
}
while self.at < total {
let entry = match self.stream.toc_row(self.at) {
Some(&entry) => entry,
None => match self.stream.harvest(self.at) {
Ok(entry) => entry,
Err(error) => {
self.at = total;
return Err(error);
}
},
};
self.at += 1;
if self.pending.remove(&entry.path_hash) {
// Nothing left to look for, so the rest of the table is never read.
if self.pending.is_empty() {
self.at = total;
}
return Ok(Some(entry));
}
}
let pending = &self.pending;
let missing: Vec<_> = self
.requested
.iter()
.copied()
.filter(|hash| pending.contains(hash))
.collect();
self.missing = missing;
Ok(None)
}
}