perf: fix backpressure on readables (#399)

src/N3Store.js

@@ -4,6 +4,9 @@ import { default as N3DataFactory, termToId, termFromId } from './N3DataFactory'
 import namespaces from './IRIs';
 import { isDefaultGraph } from './N3Util';
 
+const ITERATOR = Symbol('iter');
+
+// ## Constructor
 export class N3EntityIndex {
   constructor(options = {}) {
     this._id = 1;
@@ -386,10 +389,16 @@ export default class N3Store {
   removeMatches(subject, predicate, object, graph) {
     const stream = new Readable({ objectMode: true });
 
-    stream._read = () => {
-      for (const quad of this.readQuads(subject, predicate, object, graph))
-        stream.push(quad);
-      stream.push(null);
+    const iterable = this.readQuads(subject, predicate, object, graph);
+    stream._read = size => {
+      while (size-- > 0) {
+        const { done, value } = iterable.next();
+        if (done) {
+          stream.push(null);
+          return;
+        }
+        stream.push(value);
+      }
     };
 
     return this.remove(stream);
@@ -805,10 +814,18 @@ class DatasetCoreAndReadableStream extends Readable {
     return this.filtered.size;
   }
 
-  _read() {
-    for (const quad of this)
-      this.push(quad);
-    this.push(null);
+  _read(size) {
+    if (size > 0 && !this[ITERATOR])
+      this[ITERATOR] = this[Symbol.iterator]();
+    const iterable = this[ITERATOR];
+    while (size-- > 0) {
+      const { done, value } = iterable.next();
+      if (done) {
+        this.push(null);
+        return;
+      }
+      this.push(value);
+    }
   }
 
   add(quad) {

test/N3Store-test.js

@@ -648,6 +648,14 @@ describe('Store', () => {
       });
 
       describe('with an existing subject parameter', () => {
+        const largeStore = new Store([]);
+        const results = [];
+        for (let i = 0; i < 100; i += 1) {
+          largeStore.add(new Quad(`s${i}`, 'p1', 'o1'));
+          largeStore.add(new Quad(`s${i}`, 'p2', 'o1'));
+          results.push([`s${i}`, 'p2', 'o1']);
+        }
+
         it(
           'should return all items with this subject in all graphs',
           forResultStream(shouldIncludeAll, store.match(new NamedNode('s1'), null, null),
@@ -657,6 +665,12 @@ describe('Store', () => {
             ['s1', 'p1', 'o1', 'c4'])
         );
 
+        it(
+          'should return all items with this subject in all graphs in a large store',
+          forResultStream(shouldIncludeAll, largeStore.match(null, new NamedNode('p2'), null),
+            ...results)
+        );
+
         it('should return an object implementing the DatasetCore interface', () => {
           const subject = new NamedNode('s1');
           const dataset = store.match(subject, null, null);