1use std::fs::File;
2use std::io::{self, Read, Seek, SeekFrom};
3use std::sync::Arc;
4
5pub trait SegmentReadAt: Send + Sync {
10 fn read_at(&self, buffer: &mut [u8], offset: u64) -> io::Result<usize>;
12 fn len(&self) -> io::Result<u64>;
14 fn is_empty(&self) -> io::Result<bool> {
16 self.len().map(|length| length == 0)
17 }
18}
19
20#[derive(Clone)]
25pub struct SegmentSource {
26 backing: Arc<dyn SegmentReadAt>,
27 base: u64,
28 length: u64,
29}
30
31impl std::fmt::Debug for SegmentSource {
32 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33 formatter
34 .debug_struct("SegmentSource")
35 .field("base", &self.base)
36 .field("length", &self.length)
37 .finish_non_exhaustive()
38 }
39}
40
41impl SegmentSource {
42 pub fn from_backing(backing: Arc<dyn SegmentReadAt>) -> io::Result<Self> {
44 let length = backing.len()?;
45 Ok(Self {
46 backing,
47 base: 0,
48 length,
49 })
50 }
51
52 pub fn from_bytes(bytes: impl Into<Arc<[u8]>>) -> Self {
54 let bytes = bytes.into();
55 Self {
56 length: bytes.len() as u64,
57 base: 0,
58 backing: Arc::new(MemoryBacking(bytes)),
59 }
60 }
61
62 pub fn from_file(file: File) -> io::Result<Self> {
65 #[cfg(any(unix, windows))]
66 let backing = FileBacking(file);
67 #[cfg(not(any(unix, windows)))]
68 let backing = FileBacking(std::sync::Mutex::new(file));
69 Self::from_backing(Arc::new(backing))
70 }
71
72 pub fn subrange(&self, offset: u64, length: u64) -> io::Result<Self> {
74 if offset
75 .checked_add(length)
76 .is_none_or(|end| end > self.length)
77 {
78 return Err(io::Error::new(
79 io::ErrorKind::InvalidInput,
80 "segment subrange exceeds backing",
81 ));
82 }
83 Ok(Self {
84 backing: Arc::clone(&self.backing),
85 base: self.base + offset,
86 length,
87 })
88 }
89
90 pub fn len(&self) -> u64 {
92 self.length
93 }
94
95 pub fn is_empty(&self) -> bool {
97 self.length == 0
98 }
99
100 pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> io::Result<usize> {
102 let count = self.length.saturating_sub(offset).min(buffer.len() as u64) as usize;
103 if count == 0 {
104 return Ok(0);
105 }
106 let read = self
107 .backing
108 .read_at(&mut buffer[..count], self.base + offset)?;
109 if read > count {
110 return Err(io::Error::new(
111 io::ErrorKind::InvalidData,
112 "segment backing over-reported read length",
113 ));
114 }
115 Ok(read)
116 }
117
118 pub(crate) fn read_exact_at(&self, buffer: &mut [u8], mut offset: u64) -> io::Result<()> {
119 let mut remaining = buffer;
120 while !remaining.is_empty() {
121 match self.read_at(remaining, offset) {
122 Ok(0) => {
123 return Err(io::Error::new(
124 io::ErrorKind::UnexpectedEof,
125 "segment backing ended before requested range",
126 ));
127 }
128 Ok(count) => {
129 offset += count as u64;
130 remaining = &mut remaining[count..];
131 }
132 Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
133 Err(error) => return Err(error),
134 }
135 }
136 Ok(())
137 }
138
139 pub(crate) fn cursor(&self) -> SourceCursor {
140 SourceCursor {
141 source: self.clone(),
142 position: 0,
143 }
144 }
145}
146
147struct MemoryBacking(Arc<[u8]>);
148
149impl SegmentReadAt for MemoryBacking {
150 fn len(&self) -> io::Result<u64> {
151 Ok(self.0.len() as u64)
152 }
153 fn read_at(&self, buffer: &mut [u8], offset: u64) -> io::Result<usize> {
154 let start = usize::try_from(offset)
155 .unwrap_or(usize::MAX)
156 .min(self.0.len());
157 let count = buffer.len().min(self.0.len() - start);
158 buffer[..count].copy_from_slice(&self.0[start..start + count]);
159 Ok(count)
160 }
161}
162
163#[cfg(any(unix, windows))]
164struct FileBacking(File);
165#[cfg(not(any(unix, windows)))]
166struct FileBacking(std::sync::Mutex<File>);
167
168impl SegmentReadAt for FileBacking {
169 fn len(&self) -> io::Result<u64> {
170 #[cfg(any(unix, windows))]
171 {
172 self.0.metadata().map(|metadata| metadata.len())
173 }
174 #[cfg(not(any(unix, windows)))]
175 {
176 self.0
177 .lock()
178 .map_err(|_| io::Error::other("file backing lock poisoned"))?
179 .metadata()
180 .map(|metadata| metadata.len())
181 }
182 }
183
184 fn read_at(&self, buffer: &mut [u8], offset: u64) -> io::Result<usize> {
185 #[cfg(unix)]
186 {
187 std::os::unix::fs::FileExt::read_at(&self.0, buffer, offset)
188 }
189 #[cfg(windows)]
190 {
191 std::os::windows::fs::FileExt::seek_read(&self.0, buffer, offset)
192 }
193 #[cfg(not(any(unix, windows)))]
194 {
195 let mut file = self
196 .0
197 .lock()
198 .map_err(|_| io::Error::other("file backing lock poisoned"))?;
199 file.seek(SeekFrom::Start(offset))?;
200 file.read(buffer)
201 }
202 }
203}
204
205pub(crate) struct SourceCursor {
206 source: SegmentSource,
207 position: u64,
208}
209
210impl Read for SourceCursor {
211 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
212 let count = self.source.read_at(buffer, self.position)?;
213 self.position += count as u64;
214 Ok(count)
215 }
216}
217
218impl Seek for SourceCursor {
219 fn seek(&mut self, seek: SeekFrom) -> io::Result<u64> {
220 let position = match seek {
221 SeekFrom::Start(offset) => Some(offset),
222 SeekFrom::Current(offset) => self.position.checked_add_signed(offset),
223 SeekFrom::End(offset) => self.source.len().checked_add_signed(offset),
224 }
225 .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "segment seek out of range"))?;
226 self.position = position;
227 Ok(position)
228 }
229}