rust_ipfs/unixfs/
get.rs

1use connexa::prelude::PeerId;
2use either::Either;
3use futures::stream::BoxStream;
4use futures::{FutureExt, Stream, StreamExt, future::BoxFuture, stream::FusedStream};
5#[allow(unused_imports)]
6use rust_unixfs::walk::{ContinuedWalk, Walker};
7use std::pin::Pin;
8use std::task::Context;
9use std::{
10    path::{Path, PathBuf},
11    task::Poll,
12    time::Duration,
13};
14#[cfg(not(target_arch = "wasm32"))]
15use tokio::io::{AsyncWriteExt, BufWriter};
16use tracing::{Instrument, Span};
17
18use crate::repo::DefaultStorage;
19use crate::{Ipfs, IpfsPath, dag::IpldDag, repo::Repo};
20
21#[allow(unused_imports)]
22use super::{TraversalFailed, UnixfsStatus};
23
24#[must_use = "does nothing unless you `.await` or poll the stream"]
25pub struct UnixfsGet {
26    core: Option<Either<Ipfs, Repo<DefaultStorage>>>,
27    dest: PathBuf,
28    span: Span,
29    path: Option<IpfsPath>,
30    providers: Vec<PeerId>,
31    local_only: bool,
32    timeout: Option<Duration>,
33    stream: Option<BoxStream<'static, UnixfsStatus>>,
34}
35
36impl UnixfsGet {
37    pub fn with_ipfs(ipfs: &Ipfs, path: impl Into<IpfsPath>, dest: impl AsRef<Path>) -> Self {
38        Self::with_either(Either::Left(ipfs.clone()), path, dest)
39    }
40
41    pub fn with_repo(
42        repo: &Repo<DefaultStorage>,
43        path: impl Into<IpfsPath>,
44        dest: impl AsRef<Path>,
45    ) -> Self {
46        Self::with_either(Either::Right(repo.clone()), path, dest)
47    }
48
49    fn with_either(
50        core: Either<Ipfs, Repo<DefaultStorage>>,
51        path: impl Into<IpfsPath>,
52        dest: impl AsRef<Path>,
53    ) -> Self {
54        let path = path.into();
55        let dest = dest.as_ref().to_path_buf();
56        Self {
57            core: Some(core),
58            dest,
59            path: Some(path),
60            span: Span::current(),
61            providers: Vec::new(),
62            local_only: false,
63            timeout: None,
64            stream: None,
65        }
66    }
67
68    pub fn span(mut self, span: Span) -> Self {
69        self.span = span;
70        self
71    }
72
73    pub fn provider(mut self, peer_id: PeerId) -> Self {
74        if !self.providers.contains(&peer_id) {
75            self.providers.push(peer_id);
76        }
77        self
78    }
79
80    pub fn providers(mut self, list: &[PeerId]) -> Self {
81        self.providers = list.to_vec();
82        self
83    }
84
85    pub fn timeout(mut self, timeout: Duration) -> Self {
86        self.timeout = Some(timeout);
87        self
88    }
89
90    pub fn local(mut self) -> Self {
91        self.local_only = true;
92        self
93    }
94
95    pub fn set_local(mut self, local: bool) -> Self {
96        self.local_only = local;
97        self
98    }
99}
100
101impl Stream for UnixfsGet {
102    type Item = UnixfsStatus;
103    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
104        if self.core.is_none() && self.stream.is_none() {
105            return Poll::Ready(None);
106        }
107        loop {
108            match &mut self.stream {
109                None => {
110                    let (repo, dag) = match self.core.take().expect("ipfs or repo is used") {
111                        Either::Left(ipfs) => (ipfs.repo().clone(), ipfs.dag()),
112                        Either::Right(repo) => (repo.clone(), IpldDag::from(repo.clone())),
113                    };
114
115                    let path = self.path.take().expect("starting point exist");
116                    let providers = std::mem::take(&mut self.providers);
117                    let local_only = self.local_only;
118                    let timeout = self.timeout;
119                    let dest = self.dest.clone();
120
121                    #[cfg(not(target_arch = "wasm32"))]
122                    let stream = async_stream::stream! {
123
124                        let mut cache = None;
125                        let mut total_size = None;
126                        let mut written = 0;
127
128                        let file = match tokio::fs::File::create(dest)
129                            .await
130                            .map_err(TraversalFailed::Io) {
131                                Ok(f) => f,
132                                Err(e) => {
133                                    yield UnixfsStatus::FailedStatus { written, total_size, error: e.into() };
134                                    return;
135                                }
136                            };
137                        let mut file = BufWriter::new(file);
138
139                        let block  = match dag
140                            ._resolve(path.clone(), true, &providers, local_only, timeout)
141                            .await
142                            .map_err(TraversalFailed::Resolving)
143                            .and_then(|(resolved, _)| resolved.into_unixfs_block().map_err(TraversalFailed::Path)) {
144                                Ok(block) => block,
145                                Err(e) => {
146                                    yield UnixfsStatus::FailedStatus { written, total_size, error: e.into() };
147                                    return;
148                                }
149                        };
150
151                        let cid = block.cid();
152                        let root_name = block.cid().to_string();
153
154                        let mut walker = Walker::new(*cid, root_name);
155
156                        let mut prefetcher = super::prefetch::BlockPrefetcher::new(
157                            repo,
158                            providers,
159                            local_only,
160                            timeout,
161                            super::prefetch::WINDOW,
162                        );
163
164                        while walker.should_continue() {
165                            let next = {
166                                let (next, rest) = walker.pending_links();
167                                let next = *next;
168                                prefetcher.want(std::iter::once(next).chain(rest.copied()));
169                                next
170                            };
171                            let block = match prefetcher.get(next).await {
172                                Ok(block) => block,
173                                Err(e) => {
174                                    yield UnixfsStatus::FailedStatus { written, total_size, error: e };
175                                    return;
176                                }
177                            };
178                            let block_data = block.data();
179
180                            match walker.next(block_data, &mut cache) {
181                                Ok(ContinuedWalk::Bucket(..)) => {}
182                                Ok(ContinuedWalk::File(segment, _, _, _, size)) => {
183
184                                    if segment.is_first() {
185                                        total_size = Some(size as usize);
186                                        yield UnixfsStatus::ProgressStatus { written, total_size };
187                                    }
188                                    // even if the largest of files can have 256 kB blocks and about the same
189                                    // amount of content, try to consume it in small parts not to grow the buffers
190                                    // too much.
191
192                                    let mut n = 0usize;
193                                    let slice = segment.as_ref();
194                                    let total = slice.len();
195
196                                    while n < total {
197                                        let next = &slice[n..];
198                                        n += next.len();
199                                        if let Err(e) = file.write_all(next).await {
200                                            yield UnixfsStatus::FailedStatus { written, total_size, error: e.into() };
201                                            return;
202                                        }
203
204                                        written += n;
205                                    }
206
207                                    yield UnixfsStatus::ProgressStatus { written, total_size };
208
209                                },
210                                Ok(ContinuedWalk::Directory( .. )) | Ok(ContinuedWalk::RootDirectory( .. )) => {}, //TODO
211                                Ok(ContinuedWalk::Symlink( .. )) => {},
212                                Err(e) => {
213                                    yield UnixfsStatus::FailedStatus { written, total_size, error: e.into() };
214                                    return;
215                                }
216                            };
217                        };
218
219                        if let Err(e) = file.flush().await {
220                            yield UnixfsStatus::FailedStatus { written, total_size, error: e.into() };
221                            return;
222                        }
223                        if let Err(e) = file.get_ref().sync_all().await {
224                            yield UnixfsStatus::FailedStatus { written, total_size, error: e.into() };
225                            return;
226                        }
227
228                        yield UnixfsStatus::CompletedStatus { path, written, total_size }
229                    };
230
231                    #[cfg(target_arch = "wasm32")]
232                    let stream = async_stream::stream! {
233                        _ = repo;
234                        _ = dag;
235                        _ = path;
236                        _ = providers;
237                        _ = local_only;
238                        _ = timeout;
239                        _ = dest;
240                        yield UnixfsStatus::FailedStatus { written: 0, total_size: None, error: anyhow::anyhow!("unimplemented") };
241                    };
242
243                    self.stream = Some(stream.boxed());
244                }
245                Some(stream) => match futures::ready!(stream.poll_next_unpin(cx)) {
246                    Some(item) => {
247                        if matches!(
248                            item,
249                            UnixfsStatus::FailedStatus { .. }
250                                | UnixfsStatus::CompletedStatus { .. }
251                        ) {
252                            self.stream.take();
253                        }
254                        return Poll::Ready(Some(item));
255                    }
256                    None => {
257                        self.stream.take();
258                        return Poll::Ready(None);
259                    }
260                },
261            }
262        }
263    }
264}
265
266impl std::future::IntoFuture for UnixfsGet {
267    type Output = Result<(), anyhow::Error>;
268
269    type IntoFuture = BoxFuture<'static, Self::Output>;
270
271    fn into_future(mut self) -> Self::IntoFuture {
272        let span = self.span.clone();
273        async move {
274            while let Some(status) = self.next().await {
275                match status {
276                    UnixfsStatus::FailedStatus { error, .. } => {
277                        return Err(error);
278                    }
279                    UnixfsStatus::CompletedStatus { .. } => return Ok(()),
280                    _ => {}
281                }
282            }
283            Err::<_, anyhow::Error>(anyhow::anyhow!("Unable to get file"))
284        }
285        .instrument(span)
286        .boxed()
287    }
288}
289
290impl FusedStream for UnixfsGet {
291    fn is_terminated(&self) -> bool {
292        self.stream.is_none() && self.core.is_none()
293    }
294}
rust_ipfs/unixfs/get.rs

rust_ipfs/unixfs/
get.rs
