Add support for executing produced SPARQL queries over HTTP

processors.ttl

@@ -80,6 +80,12 @@ js:SPARQLIngest a js:JsProcess;
     sh:name "transactionConfig";
     sh:maxCount 1;
     sh:minCount 0;
+  ], [
+    sh:path js:graphStoreUrl;
+    sh:datatype xsd:string;
+    sh:name "graphStoreUrl";
+    sh:maxCount 1;
+    sh:minCount 0;
   ].
 
 [ ] a sh:NodeShape;

src/SPARQLIngest.ts

@@ -3,6 +3,7 @@ import { SDS } from "@treecg/types";
 import { Store, Parser, DataFactory } from "n3";
 import { CREATE, UPDATE, DELETE } from "./SPARQLQueries";
 import { Quad_Subject, Term } from "@rdfjs/types";
+import { doSPARQLRequest } from "./Utils";
 
 const { quad, namedNode } = DataFactory;
 
@@ -24,6 +25,7 @@ export type IngestConfig = {
     changeSemantics?: ChangeSemantics;
     targetNamedGraph?: string;
     transactionConfig?: TransactionConfig;
+    graphStoreUrl?: string;
 };
 
 export type TransactionMember = {
@@ -35,7 +37,7 @@ export type TransactionMember = {
 export async function sparqlIngest(
     memberStream: Stream<string>,
     config: IngestConfig,
-    sparqlWriter: Writer<string>
+    sparqlWriter?: Writer<string>
 ) {
     let transactionMembers: TransactionMember[] = [];
 
@@ -146,14 +148,24 @@ export async function sparqlIngest(
 
             // Execute the update query
             if (query) {
-                await sparqlWriter.push(query);
-            }
+                const outputPromises = [];
+                if (sparqlWriter) {
+                    outputPromises.push(sparqlWriter.push(query));
+                }
+                if (config.graphStoreUrl) {
+                    outputPromises.push(doSPARQLRequest(query, config.graphStoreUrl));
+                }
+
+                await Promise.all(outputPromises);
+            } 
         } else {
             throw new Error(`[sparqlIngest] No member IRI found in received RDF data: \n${rawQuads}`);
         }
     });
 
-    memberStream.on("end", async () => await sparqlWriter.end());
+    if (sparqlWriter) {
+        memberStream.on("end", async () => await sparqlWriter.end());
+    }
 }
 
 function verifyTransaction(stores: Store[], transactionIdPath: string, transactionId: Term): void {

src/Utils.ts

@@ -0,0 +1,14 @@
+
+export async function doSPARQLRequest(query: string, url: string): Promise<void> {
+    const res = await fetch(url, {
+        method: "POST",
+        headers: {
+            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
+        },
+        body: `query=${encodeURIComponent(query)}`
+    });
+
+    if (!res.ok) {
+        throw new Error(`HTTP request failed with code ${res.status} and message: \n${await res.text()}`);
+    }
+}

tests/processors.test.ts

@@ -40,7 +40,8 @@ describe("Tests for SPARQL ingest processor", async () => {
                     js:transactionConfig [
                         js:transactionIdPath "http://ex.org/transactionId";
                         js:transactionEndPath "http://ex.org/transactionEnd"
-                    ]
+                    ];
+                    js:graphStoreUrl "http://ex.org/myGraphStore"
                 ];
                 js:sparqlWriter <jw>.
         `;
@@ -73,6 +74,7 @@ describe("Tests for SPARQL ingest processor", async () => {
         expect(ingestConfig.targetNamedGraph).toBe("http://ex.org/myGraph");
         expect(ingestConfig.transactionConfig.transactionIdPath).toBe("http://ex.org/transactionId");
         expect(ingestConfig.transactionConfig.transactionEndPath).toBe("http://ex.org/transactionEnd");
+        expect(ingestConfig.graphStoreUrl).toBe("http://ex.org/myGraphStore");
         testWriter(sparqlWriter);
 
         await checkProc(env.file, env.func);