Extract hextuple processor from Libro

package.json

@@ -43,6 +43,7 @@
     "@ontologies/schema": ">=1.0.0",
     "@ontologies/shacl": ">=1.0.0",
     "@ontologies/xsd": ">=1.0.0",
+    "can-ndjson-stream": "^1.0.2",
     "http-status-codes": ">= 1.x",
     "n-quads-parser": "^2.1.0-3"
   },

src/transformers/hextuples.ts

@@ -0,0 +1,23 @@
+import { Quad } from "@ontologies/core";
+
+import { LinkedRenderStore } from "../LinkedRenderStore";
+import {
+  ResponseAndFallbacks,
+  ResponseTransformer,
+} from "../types";
+import { hextupleTransformer } from "../utilities/hextupleProcessor";
+
+export const hextupleProcessor = {
+  acceptValue: 1.0,
+  mediaTypes: ["application/hex+x-ndjson"],
+
+  transformer: (store: LinkedRenderStore<any>): ResponseTransformer => (res: ResponseAndFallbacks): Promise<Quad[]> => {
+    const isExpedited = res.hasOwnProperty("expedite")
+      ? (res as any).expedite
+      : false;
+
+    return hextupleTransformer(res)
+      .then((delta) => store.queueDelta(delta, isExpedited))
+      .then(() => []);
+  },
+};

src/transformers/index.ts

@@ -1,7 +1,9 @@
 import { linkedDeltaProcessor } from "./linked-delta";
+import { hextupleProcessor } from "./hextuples";
 import { createProcessRDF } from "./rdf-formats-common";
 
 export const transformers = {
     createProcessRDF,
+    hextupleProcessor,
     linkedDeltaProcessor,
 };

src/utilities/hextupleProcessor.ts

@@ -0,0 +1,113 @@
+import rdf, { BlankNode, Quad, SomeTerm } from "@ontologies/core";
+import * as rdfx from "@ontologies/rdf";
+// @ts-ignore
+import NdjsonStream from "can-ndjson-stream";
+
+import {
+  ExtensionResponse,
+  RDFLibFetcherResponse,
+  ResponseAndFallbacks,
+  ResponseTransformer,
+} from "../types";
+
+enum HexPosition {
+  Subject = 0,
+  Predicate,
+  Value,
+  Datatype,
+  Language,
+  Graph,
+}
+
+let hasReadableStreamConstructor = false;
+
+try {
+  // tslint:disable-next-line:typedef no-empty no-unused-expression
+  new ReadableStream({ start() {} });
+  hasReadableStreamConstructor = true;
+} catch (e) {
+  // ignore
+}
+
+export const hextupleTransformer: ResponseTransformer = async (res: ResponseAndFallbacks): Promise<Quad[]> => {
+  const bnMap: { [s: string]: BlankNode } = {};
+  // Skip the (expensive) proxy object when parsing
+  const quad = rdf.quad.bind(rdf);
+  const literal = rdf.literal.bind(rdf);
+  const namedNode = rdf.namedNode.bind(rdf);
+  const blankNodeF = rdf.blankNode.bind(rdf);
+  const defaultGraph = rdf.defaultGraph.bind(rdf);
+
+  const blankNode = (v: string): BlankNode => {
+    if (!bnMap[v]) {
+      bnMap[v] = blankNodeF();
+    }
+
+    return bnMap[v];
+  };
+
+  const object = (v: string, dt: string, l: string): SomeTerm => {
+    if (l) {
+      return literal(v, l);
+    } else if (dt === rdfx.ns("namedNode").value) {
+      return namedNode(v);
+    } else if (dt === rdfx.ns("blankNode").value) {
+      return blankNode(v);
+    }
+
+    return literal(v, namedNode(dt));
+  };
+
+  const lineToQuad = (h: string[]): Quad => quad(
+    h[HexPosition.Subject].startsWith("_:") ? blankNode(h[HexPosition.Subject]) : namedNode(h[HexPosition.Subject]),
+    namedNode(h[HexPosition.Predicate]),
+    object(h[HexPosition.Value], h[HexPosition.Datatype], h[HexPosition.Language]),
+    h[HexPosition.Graph] ? namedNode(h[HexPosition.Graph]) : defaultGraph(),
+  );
+
+  const delta: Quad[] = [];
+  let parse;
+
+  if (res instanceof Response && hasReadableStreamConstructor) {
+    const stream = new NdjsonStream(res.body);
+    const reader = stream.getReader();
+
+    let read: any;
+    parse = reader
+      .read()
+      .then(read = (result: { done: boolean, value: string[] }) => {
+        if (result.done) {
+          return;
+        }
+
+        delta.push(lineToQuad(result.value));
+
+        return reader.read().then(read);
+      });
+  } else {
+    let body;
+
+    if (res instanceof Response) {
+      body = res.text();
+    } else if (typeof XMLHttpRequest !== "undefined"
+      && res instanceof XMLHttpRequest
+      || typeof (res as any).responseText === "string") {
+      body = Promise.resolve((res as XMLHttpRequest | RDFLibFetcherResponse).responseText);
+    } else {
+      body = Promise.resolve((res as ExtensionResponse).body);
+    }
+
+    parse = body
+      .then((text) => {
+        for (const line of text.split("\n")) {
+          if (line.length > 0) {
+            delta.push(lineToQuad(JSON.parse(line)));
+          }
+        }
+      });
+  }
+
+  await parse;
+
+  return delta;
+};

yarn.lock

@@ -2219,6 +2219,18 @@ camelcase@^6.2.0:
   resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-6.2.0.tgz#924af881c9d525ac9d87f40d964e5cea982a1809"
   integrity sha512-c7wVvbw3f37nuobQNtgsgG9POC9qMbNuMQmTCqZv23b6MIz0fcYpBiOlv9gEN/hdLdnZTDQhg6e9Dq5M1vKvfg==
 
+can-namespace@^1.0.0:
+  version "1.0.0"
+  resolved "https://registry.yarnpkg.com/can-namespace/-/can-namespace-1.0.0.tgz#0b8fafafbb11352b9ead4222ffe3822405b43e99"
+  integrity sha1-C4+vr7sRNSuerUIi/+OCJAW0Ppk=
+
+can-ndjson-stream@^1.0.2:
+  version "1.0.2"
+  resolved "https://registry.yarnpkg.com/can-ndjson-stream/-/can-ndjson-stream-1.0.2.tgz#6a8131f9c8c697215163b3fe49a0c02e4439cb47"
+  integrity sha512-//tM8wcTV42SyD1JGua7WMVftZEeTwapcHJTTe3vJwuVywXD01CJbdEkgwRYjy2evIByVJV21ZKBdSv5ygIw1w==
+  dependencies:
+    can-namespace "^1.0.0"
+
 caniuse-lite@^1.0.30001173:
   version "1.0.30001180"
   resolved "https://registry.yarnpkg.com/caniuse-lite/-/caniuse-lite-1.0.30001180.tgz#67abcd6d1edf48fa5e7d1e84091d1d65ab76e33b"