RDF Streaming Store

A read-only RDF/JS store that allows parallel data lookup and insertion. It works in both JavaScript and TypeScript.

Concretely, this means that match() calls happening before import() calls, will still consider those triples that are inserted later, which is done by keeping the response streams of match() open. Only when the end() method is invoked, all response streams will close, and the StreamingStore will be considered immutable.

WARNING: end() MUST be called at some point, otherwise all match streams will remain unended.

If using TypeScript, it is recommended to use this in conjunction with @rdfjs/types.

Installation

1$ npm install rdf-streaming-store

or

1$ yarn add rdf-streaming-store

This package also works out-of-the-box in browsers via tools such as webpack and browserify.

Usage

A new StreamingStore can be created as follows:

1import { StreamingStore } from 'rdf-streaming-store';
2
3const store = new StreamingStore();

Inserting quads

Following the RDF/JS Sink interface, new quads can be added using the import method, which accepts a stream of quads:

1const quad = require('rdf-quad');
2const streamifyArray = require('streamify-array');
3
4// Somehow create a quad stream
5const quadStream = streamifyArray([
6  quad('s3', 'p3', 'o3'),
7  quad('s4', 'p4', 'o4'),
8]);
9
10// Import it into the store
11store.import(quadStream);

After inserting your quads, you MUST call end() to make sure that match calls will end their response streams:

1store.end();

After calling end(), importing new quads is not allowed.

Finding quads

Following the RDF/JS Source interface, quads can be found using the match method, which returns a stream of quads:

1import type * as RDF from '@rdfjs/types';
2import { DataFactory } from 'rdf-data-factory';
3
4const DF = new DataFactory();
5const returnStream = store.match(undefined, DF.namedNode('p3'), DF.namedNode('o3'), undefined);
6
7returnStream.on('data', (quad: RDF.Quad) => {
8  console.log(quad);
9});
10returnStream.on('error', (error) => {
11  console.log(error);
12});
13returnStream.on('end', () => {
14  console.log('Done!');
15});

Note that the returnStream will not end until the store.end() has been invoked.

match() can be called before import()

Since match() calls will only end after calling end(), quads that are imported after initiating the match() call, can still be emitted in the created match() stream.

1const store = new StreamingStore();
2const returnStream = store.match(undefined, DF.namedNode('p3'), DF.namedNode('o3'), undefined);
3
4returnStream.on('data', (quad: RDF.Quad) => {
5  console.log(quad);
6});
7returnStream.on('end', () => {
8  console.log('Done!');
9});
10
11// At this stage, the store is empty, so no quads will be printed yet
12
13// After importing some quads into the store, the s3-p3-o3 triple will be printed
14store.import(streamifyArray([
15  quad('s3', 'p3', 'o3'),
16  quad('s4', 'p4', 'o4'),
17]));
18
19// After importing some more triples, another triple will be printed
20store.import(streamifyArray([
21  quad('sOther', 'p3', 'o3'),
22]));
23
24// Since we mark the store as ended, the returnStream will print `Done!`
25store.end();

Determining if the store has ended

1const store = new StreamingStore();
2
3store.hasEnded(); // Will return false
4store.addEndListener(() => {
5  console.log("store has ended");
6});
7
8store.addEndListener(() => {
9  console.log("store has ended second call");
10});
11
12
13store.end(); // All the listeners will be called in the order they have been added
14
15store.hasEnded(); // Will return true

License

This software is written by Maarten Vandenbrande and Ruben Taelman.

This code is released under the MIT license.