Incremunica 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 @incremunica/streaming-store

or

1$ yarn add @incremunica/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 '@incremunica/streaming-store';
2
3const store = new StreamingStore();

Inserting/removing 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]);
9store.import(quadStream); // Import it into the store
10
11store.addQuad(quad('s5', 'p5', 'o5')); // Singular quad insertions
12
13// Somehow create a quad stream
14const otherQuadStream = streamifyArray([
15  quad('s3', 'p3', 'o3'),
16]);
17store.remove(otherQuadStream); // Remove it from the store
18
19store.removeQuad(quad('s4', 'p4', 'o4')); // singular quad deletions
20
21const IncrementalQuad = require("@incremunica/types").Quad;
22let deletionQuad = <IncrementalQuad> quad('s5', 'p5', 'o5'); // Make an incremental quad
23deletionQuad.diff = false; // Set diff as false (marks the quad as deleted)
24store.addQuad(deletionQuad); // Will remove the quad from the store

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();

License

This software is written by Maarten Vandenbrande and Ruben Taelman.

This code is released under the MIT license.