https://comunica.dev/js/comunica-worker.js

1/* Runs the Comunica query engine off the main thread, for the demo on the home page. */
2
3const ENGINE_URL = 'https://rdf.js.org/comunica-browser/versions/v4/engines/query-sparql/comunica-browser.js';
4const PROGRESS_INTERVAL = 250;
5
6let engine;
7let active;
8let requests = 0;
9
10// Wrap fetch before the engine is loaded, so that all of its requests are counted.
11const originalFetch = self.fetch.bind(self);
12self.fetch = function (...args) {
13  requests++;
14  return originalFetch(...args);
15};
16
17self.onmessage = function (event) {
18  if (event.data.type === 'query') {
19    runQuery(event.data);
20  } else if (event.data.type === 'stop') {
21    stopQuery(true);
22  } else if (event.data.type === 'preload') {
23    preloadEngine();
24  }
25};
26
27// Fetch the engine ahead of the first query, without contacting any source.
28function preloadEngine() {
29  try {
30    loadEngine();
31    postMessage({ type: 'ready' });
32  } catch (error) {
33    postMessage({ type: 'engineError', message: String((error && error.message) || error) });
34  }
35}
36
37// Load the pre-built engine on first use, and keep it for later queries.
38function loadEngine() {
39  if (!engine) {
40    importScripts(ENGINE_URL);
41    engine = new self.Comunica.QueryEngine();
42  }
43  return engine;
44}
45
46function termToJson(term) {
47  return term ? { termType: term.termType, value: term.value } : null;
48}
49
50// Finish the active query, ignoring any later events of the same run.
51function finish(run, truncated) {
52  if (run.finished) {
53    return;
54  }
55  run.finished = true;
56  clearInterval(run.timer);
57  if (run.iterator) {
58    run.iterator.destroy();
59  }
60  if (active === run) {
61    active = undefined;
62  }
63  postMessage({ type: 'end', token: run.token, requests, truncated });
64}
65
66function fail(run, error) {
67  if (run.finished) {
68    return;
69  }
70  run.finished = true;
71  clearInterval(run.timer);
72  if (active === run) {
73    active = undefined;
74  }
75  postMessage({ type: 'error', token: run.token, message: String((error && error.message) || error), requests });
76}
77
78function stopQuery(notify) {
79  if (active) {
80    const run = active;
81    if (notify) {
82      finish(run, false);
83    } else {
84      run.finished = true;
85      clearInterval(run.timer);
86      if (run.iterator) {
87        run.iterator.destroy();
88      }
89      active = undefined;
90    }
91  }
92}
93
94async function runQuery({ token, query, sources, limit }) {
95  stopQuery(false);
96  requests = 0;
97
98  const run = { token, finished: false, iterator: undefined, timer: undefined };
99  active = run;
100  run.timer = setInterval(() => postMessage({ type: 'progress', token, requests }), PROGRESS_INTERVAL);
101
102  try {
103    const result = await loadEngine().query(query, { sources });
104    if (run.finished) {
105      return;
106    }
107
108    // Queries that produce a single answer instead of a stream of rows.
109    if (result.resultType === 'boolean' || result.resultType === 'void') {
110      const value = await result.execute();
111      postMessage({ type: 'variables', token, variables: ['result'] });
112      postMessage({
113        type: 'result',
114        token,
115        row: [{ termType: 'Literal', value: result.resultType === 'boolean' ? String(value) : 'Done' }],
116        requests,
117      });
118      finish(run, false);
119      return;
120    }
121
122    const quads = result.resultType === 'quads';
123    let variables;
124    if (quads) {
125      variables = ['subject', 'predicate', 'object'];
126    } else {
127      const metadata = await result.metadata();
128      // Comunica reports variables either directly or wrapped, depending on the version.
129      variables = (metadata.variables || []).map(entry => (entry.variable ? entry.variable.value : entry.value));
130    }
131    if (run.finished) {
132      return;
133    }
134    postMessage({ type: 'variables', token, variables });
135
136    run.iterator = await result.execute();
137    let count = 0;
138    run.iterator.on('data', (item) => {
139      count++;
140      // One result past the limit proves that more exist, without rendering them.
141      if (count > limit) {
142        finish(run, true);
143        return;
144      }
145      const row = quads
146        ? [termToJson(item.subject), termToJson(item.predicate), termToJson(item.object)]
147        : variables.map(variable => termToJson(item.get(variable)));
148      postMessage({ type: 'result', token, row, requests });
149    });
150    run.iterator.on('end', () => finish(run, false));
151    run.iterator.on('error', error => fail(run, error));
152  } catch (error) {
153    fail(run, error);
154  }
155}