Fetch a bounded byte range from `store` and return it as a stream
(
store: Arc<dyn ObjectStore>,
location: object_store::path::Path,
range: std::ops::Range<u64>,
)
| 85 | |
| 86 | /// Fetch a bounded byte range from `store` and return it as a stream |
| 87 | async fn get_stream( |
| 88 | store: Arc<dyn ObjectStore>, |
| 89 | location: object_store::path::Path, |
| 90 | range: std::ops::Range<u64>, |
| 91 | ) -> object_store::Result<BoxStream<'static, object_store::Result<Bytes>>> { |
| 92 | let opts = GetOptions { |
| 93 | range: Some(GetRange::Bounded(range.clone())), |
| 94 | ..Default::default() |
| 95 | }; |
| 96 | let result = store.get_opts(&location, opts).await?; |
| 97 | |
| 98 | #[cfg(not(target_arch = "wasm32"))] |
| 99 | if let GetResultPayload::File(mut file, _path) = result.payload { |
| 100 | use std::io::{Read, Seek, SeekFrom}; |
| 101 | const CHUNK_SIZE: u64 = 8 * 1024; |
| 102 | |
| 103 | file.seek(SeekFrom::Start(range.start)).map_err(|e| { |
| 104 | object_store::Error::Generic { |
| 105 | store: "local", |
| 106 | source: Box::new(e), |
| 107 | } |
| 108 | })?; |
| 109 | |
| 110 | return Ok(futures::stream::try_unfold( |
| 111 | (file, range.end - range.start), |
| 112 | move |(mut file, remaining)| async move { |
| 113 | if remaining == 0 { |
| 114 | return Ok(None); |
| 115 | } |
| 116 | let to_read = remaining.min(CHUNK_SIZE); |
| 117 | let cap = usize::try_from(to_read).map_err(|e| { |
| 118 | object_store::Error::Generic { |
| 119 | store: "local", |
| 120 | source: Box::new(e), |
| 121 | } |
| 122 | })?; |
| 123 | |
| 124 | let mut buf = Vec::with_capacity(cap); |
| 125 | let read = |
| 126 | (&mut file) |
| 127 | .take(to_read) |
| 128 | .read_to_end(&mut buf) |
| 129 | .map_err(|e| object_store::Error::Generic { |
| 130 | store: "local", |
| 131 | source: Box::new(e), |
| 132 | })?; |
| 133 | Ok(Some((Bytes::from(buf), (file, remaining - read as u64)))) |
| 134 | }, |
| 135 | ) |
| 136 | .boxed()); |
| 137 | } |
| 138 | |
| 139 | Ok(result.into_stream()) |
| 140 | } |
| 141 | |
| 142 | impl AlignedBoundaryStream { |
| 143 | /// Open a ranged byte stream from `store` and return a ready-to-poll |