fix: add tests and fixes to stream source

maartyman · Jan 20, 2025 · 1b01ad6 · 1b01ad6
1 parent 3d1fbfa
commit 1b01ad6
Show file tree

Hide file tree

Showing 10 changed files with 550 additions and 70 deletions.
diff --git a/engines/config-query-sparql-incremental/config/query-source-identify/actors/stream.json b/engines/config-query-sparql-incremental/config/query-source-identify/actors/stream.json
@@ -10,7 +10,7 @@
     {
       "@id": "urn:comunica:default:query-source-identify/actors#stream",
       "@type": "ActorQuerySourceIdentifyStream",
-      "mediatorQuerySourceIdentify": { "@id": "urn:comunica:default:query-source-identify/mediators#main" },
+      "mediatorContextPreprocess": { "@id": "urn:comunica:default:context-preprocess/mediators#main" },
       "mediatorRdfMetadataAccumulate": { "@id": "urn:comunica:default:rdf-metadata-accumulate/mediators#main" }
     }
   ]

diff --git a/engines/query-sparql-incremental/test/QuerySparql-test.ts b/engines/query-sparql-incremental/test/QuerySparql-test.ts
@@ -383,19 +383,15 @@ describe('System test: QuerySparql (without polly)', () => {
 
       const sourcesStream = new PassThrough({ objectMode: true });
       sourcesStream.push({
-        value: 'http://localhost:8787',
-        isAddition: true,
-      });
-      sourcesStream.push({
-        value: streamingStore,
+        querySource: 'http://localhost:8787',
         isAddition: true,
       });
 
       bindingStream = await engine.queryBindings(`SELECT * WHERE {
     ?s ?p ?o.
   }`, {
         // @ts-expect-error
-        sources: [ sourcesStream ],
+        sources: [ streamingStore, sourcesStream ],
         pollingPeriod: 1000,
       });
 
@@ -416,7 +412,7 @@ describe('System test: QuerySparql (without polly)', () => {
       ]).setContextEntry(KeysBindings.isAddition, true));
 
       sourcesStream.push({
-        value: 'http://localhost:8787',
+        querySource: 'http://localhost:8787',
         isAddition: false,
       });
 
@@ -563,7 +559,7 @@ describe('System test: QuerySparql (with polly)', () => {
         // @ts-expect-error
         sources: [ new ArrayIterator([
           {
-            value: 'https://www.rubensworks.net/',
+            querySource: 'https://www.rubensworks.net/',
             isAddition: true,
           },
         ]) ],

diff --git a/packages/actor-query-source-identify-stream/README.md b/packages/actor-query-source-identify-stream/README.md
@@ -24,7 +24,7 @@ After installing, this package can be added to your engine's configuration as fo
     {
       "@id": "urn:comunica:default:query-source-identify/actors#stream",
       "@type": "ActorQuerySourceIdentifyStream",
-      "mediatorQuerySourceIdentify": { "@id": "urn:comunica:default:query-source-identify/mediators#main" },
+      "mediatorContextPreprocess": { "@id": "urn:comunica:default:context-preprocess/mediators#main" },
       "mediatorRdfMetadataAccumulate": { "@id": "urn:comunica:default:rdf-metadata-accumulate/mediators#main" }
     }
   ]

diff --git a/packages/actor-query-source-identify-stream/lib/ActorQuerySourceIdentifyStream.ts b/packages/actor-query-source-identify-stream/lib/ActorQuerySourceIdentifyStream.ts
@@ -1,30 +1,31 @@
+import type { MediatorContextPreprocess } from '@comunica/bus-context-preprocess';
 import type {
   IActionQuerySourceIdentify,
   IActorQuerySourceIdentifyOutput,
   IActorQuerySourceIdentifyArgs,
-  MediatorQuerySourceIdentify,
 } from '@comunica/bus-query-source-identify';
 import { ActorQuerySourceIdentify } from '@comunica/bus-query-source-identify';
 import type { MediatorRdfMetadataAccumulate } from '@comunica/bus-rdf-metadata-accumulate';
 import { KeysInitQuery } from '@comunica/context-entries';
 import type { IActorTest, TestResult } from '@comunica/core';
 import { failTest, passTestVoid, ActionContext } from '@comunica/core';
 import type { ComunicaDataFactory } from '@comunica/types';
+import type { QuerySourceStream, QuerySourceUnidentifiedExpanded } from '@incremunica/types';
 import { StreamQuerySources } from './StreamQuerySources';
 
 /**
  * An incremunica Stream Sources Query Source Identify Actor.
  */
 export class ActorQuerySourceIdentifyStream extends ActorQuerySourceIdentify {
-  public readonly mediatorQuerySourceIdentify: MediatorQuerySourceIdentify;
   public readonly mediatorRdfMetadataAccumulate: MediatorRdfMetadataAccumulate;
+  public readonly mediatorContextPreprocess: MediatorContextPreprocess;
 
   public constructor(args: IActorQuerySourceIdentifyStreamSourcesArgs) {
     super(args);
   }
 
   public async test(action: IActionQuerySourceIdentify): Promise<TestResult<IActorTest>> {
-    const source = action.querySourceUnidentified;
+    const source = <QuerySourceUnidentifiedExpanded>action.querySourceUnidentified;
     if (source.type !== 'stream') {
       return failTest(`${this.name} requires a single query source with stream type to be present in the context.`);
     }
@@ -36,10 +37,10 @@ export class ActorQuerySourceIdentifyStream extends ActorQuerySourceIdentify {
     return {
       querySource: {
         source: new StreamQuerySources(
-          <any>action.querySourceUnidentified.value,
+          <QuerySourceStream><any>action.querySourceUnidentified.value,
           dataFactory,
-          this.mediatorQuerySourceIdentify,
           this.mediatorRdfMetadataAccumulate,
+          this.mediatorContextPreprocess,
           action.context,
         ),
         context: action.querySourceUnidentified.context ?? new ActionContext(),
@@ -50,11 +51,11 @@ export class ActorQuerySourceIdentifyStream extends ActorQuerySourceIdentify {
 
 export interface IActorQuerySourceIdentifyStreamSourcesArgs extends IActorQuerySourceIdentifyArgs {
   /**
-   * A mediator for identifying query sources.
+   * A mediator for accumulating metadata.
    */
-  mediatorQuerySourceIdentify: MediatorQuerySourceIdentify;
+  mediatorRdfMetadataAccumulate: MediatorRdfMetadataAccumulate;
   /**
-   * A mediator for identifying query sources.
+   * A mediator for preprocessing the context.
    */
-  mediatorRdfMetadataAccumulate: MediatorQuerySourceIdentify;
+  mediatorContextPreprocess: MediatorContextPreprocess;
 }