ldp 0.1.0

A library to assist with the creation and maintenance of remote RDF data via LDP
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

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};
use tracing::debug_span;

/// Recursively traverse an entire repository.
///
/// Beginning with the root URL, this [`Stream`] will fetch the resource, emit it as a
/// [`RdfSource<Dataset>`], 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<BoxStream<'a, crate::Result<RdfSource<Dataset>>>>,
    visited_urls: HashSet<Url>,
    pending_urls: HashSet<Url>,
    parallelism: Option<usize>,
    span: tracing::Span,
}

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<usize>) -> Self {
        let mut this = Self {
            http_client,
            stream: SelectAll::new(),
            visited_urls: HashSet::new(),
            pending_urls: HashSet::new(),
            parallelism,
            span: debug_span!("Traverse Repository", %root),
        };
        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<RdfSource<Dataset>>> {
        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::<HashSet<_>>();
            } else {
                next_batch = HashSet::new();
            }
        } else {
            next_batch = self.pending_urls.iter().cloned().collect::<HashSet<_>>();
        }

        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<RdfSource<Dataset>>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let _enter = self.span.clone().entered();
        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
    }
}