Fix quad event not being emitted anymore

Author: constraintAutomaton

lib/StreamingStore.ts

@@ -75,9 +75,14 @@ implements RDF.Source<Q>, RDF.Sink<RDF.Stream<Q>, EventEmitter> {
         quad.graph,
       )) {
         for (const pendingStream of this.pendingStreams.getPendingStreamsForQuad(quad)) {
+          /**
+           * The pendingStream emits 'quad' events even before it is initialized.
+           * This allows us to detect when matching quads are added to the store,
+           * without having to consume the stream returned by the `match` method.
+           */
+          pendingStream.emit('quad', quad);
           if ((<any> pendingStream).isInitialized) {
             pendingStream.push(quad);
-            pendingStream.emit('quad', quad);
           }
         }
       }
@@ -116,6 +121,9 @@ implements RDF.Source<Q>, RDF.Sink<RDF.Stream<Q>, EventEmitter> {
       const pendingStream = new PassThrough({ objectMode: true });
       this.pendingStreams.addPatternListener(pendingStream, subject, predicate, object, graph);
       stream = Readable.from(StreamingStore.concatStreams([ storeResult, pendingStream ]));
+      pendingStream.on('quad', quad => {
+        stream.emit('quad', quad);
+      });
       (<any> stream)._pipeSource = storeResult;
 
       // This is an ugly hack to annotate pendingStream with the isInitialized once the store stream started being read.

test/StreamingStore-test.ts

@@ -1,4 +1,5 @@
 import 'jest-rdf';
+import type * as RDF from '@rdfjs/types';
 import arrayifyStream from 'arrayify-stream';
 import { promisifyEventEmitter } from 'event-emitter-promisify/dist';
 import { Store } from 'n3';
@@ -555,6 +556,31 @@ describe('StreamingStore', () => {
     expect(listener).toHaveBeenCalledTimes(1);
   });
 
+  it('should emit a quad event when new matching quads are imported to the store', async() => {
+    await promisifyEventEmitter(store.import(streamifyArray([
+      quad('s1', 'p1', 'o1'),
+      quad('s2', 'p2', 'o2'),
+    ])));
+
+    const stream = store.match(DF.namedNode('s1'));
+    const quads: RDF.Quad[] = [];
+    stream.on('quad', importedQuad => {
+      quads.push(importedQuad);
+    });
+
+    await promisifyEventEmitter(store.import(streamifyArray([
+      quad('s1', 'p3', 'o3'),
+      quad('s4', 'p4', 'o4'),
+    ])));
+
+    store.end();
+    expect(quads).toStrictEqual(
+      [
+        quad('s1', 'p3', 'o3'),
+      ],
+    );
+  });
+
   it('handle the reporting of the ending of the store', () => {
     expect(store.hasEnded()).toBe(false);
     store.end();