diff --git a/ldp/src/lib.rs b/ldp/src/lib.rs index 7d8a538..4aadbbd 100644 --- a/ldp/src/lib.rs +++ b/ldp/src/lib.rs @@ -5,6 +5,7 @@ pub mod header; pub mod middleware; mod rdf_source; mod resource; +pub mod traverse; pub mod vocab; #[cfg(feature = "keyed")] diff --git a/ldp/src/traverse.rs b/ldp/src/traverse.rs new file mode 100644 index 0000000..a072595 --- /dev/null +++ b/ldp/src/traverse.rs @@ -0,0 +1,117 @@ +use crate::{RdfSource, ResourceRequestBuilder, vocab}; +use futures::stream::{BoxStream, SelectAll}; +use futures::{FutureExt, Stream, StreamExt}; +use oxigraph::model::{Dataset, TermRef}; +use reqwest_middleware::ClientWithMiddleware; +use reqwest_middleware::reqwest::Url; +use std::collections::HashSet; +use std::pin::Pin; +use std::task::{Context, Poll, ready}; + +/// Recursively traverse an entire repository. +/// +/// Beginning with the root URL, this [`Stream`] will fetch the resource, emit it as a +/// [`RdfSource`], and then fetch every child resource, as determined by the +/// [`ldp:contains`](https://www.w3.org/TR/ldp/#ldpc) predicate. +/// +/// All visited URLs are stored internally to prevent loops. +pub struct Traverse<'a> { + http_client: ClientWithMiddleware, + stream: SelectAll>>>, + visited_urls: HashSet, + pending_urls: HashSet, + parallelism: Option, +} + +impl<'a> Traverse<'a> { + /// Create a new traversal, starting with the `root`. `parallelism` limits the number of + /// requests which may be in-flight at any given time. A value of `None` indicates infinite + /// parallelism and should be used this with caution, as the reqwest library will open a + /// potentially unbounded number of TCP connections. + pub fn new(http_client: ClientWithMiddleware, root: Url, parallelism: Option) -> Self { + let mut this = Self { + http_client, + stream: SelectAll::new(), + visited_urls: HashSet::new(), + pending_urls: HashSet::new(), + parallelism, + }; + this.visited_urls.insert(root.clone()); + this.stream + .push(Self::fetch(this.http_client.clone(), root)); + this + } + + fn fetch( + http_client: ClientWithMiddleware, + url: Url, + ) -> BoxStream<'a, crate::Result>> { + async { + let resource = ResourceRequestBuilder::with_client_and_url(http_client, url) + .build() + .send() + .await?; + let rdf_source = resource.into_rdf_source().await?; + Ok(rdf_source) + } + .into_stream() + .boxed() + } + + fn push_pending(&mut self) { + let next_batch; + if let Some(parallelism) = self.parallelism { + if let Some(mut available) = parallelism.checked_sub(self.stream.len()) { + // Plus 1 to account for the document currently being processed. + available += 1; + next_batch = self + .pending_urls + .iter() + .take(available) + .cloned() + .collect::>(); + } else { + next_batch = HashSet::new(); + } + } else { + next_batch = self.pending_urls.iter().cloned().collect::>(); + } + + for url in next_batch { + self.pending_urls.remove(&url); + let client = self.http_client.clone(); + self.stream.push(Self::fetch(client, url)); + } + } +} + +impl<'a> Stream for Traverse<'a> { + type Item = crate::Result>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let result = match ready!(self.stream.poll_next_unpin(cx)) { + Some(Ok(document)) => { + self.visited_urls.insert(document.origin().clone()); + if let Some(described_by) = document.described_by() { + self.visited_urls.insert(described_by.clone()); + } + + let dataset = document.dataset(); + for quad in dataset.quads_for_predicate(vocab::ldp::CONTAINS) { + if let TermRef::NamedNode(node) = quad.object + && let Ok(url) = Url::parse(node.as_str()) + && !self.visited_urls.contains(&url) + { + self.pending_urls.insert(url); + } + } + Poll::Ready(Some(Ok(document))) + } + Some(Err(err)) => Poll::Ready(Some(Err(err))), + None => Poll::Ready(None), + }; + + self.push_pending(); + result + } +}