refactor: generator writing to graph store

Author: tpluscode

package-lock.json

@@ -0,0 +1,35 @@
+import { PassThrough } from 'readable-stream'
+
+/**
+ * @param {(stream: import('readable-stream').PassThrough) => Promise<void>} openConnection
+ * @param {object} context
+ * @param {import('winston').Logger} context.logger
+ */
+export default function (openConnection, { logger }) {
+  /**
+   * @this {import('barnard59-core').Context}
+   * @param {AsyncIterable<import('@rdfjs/types').Quad>} quads
+   */
+  return async function * (quads) {
+    const quadStream = new PassThrough({
+      objectMode: true,
+    })
+    let request
+
+    for await (const quad of quads) {
+      if (!request) {
+        request = openConnection(quadStream)
+      }
+
+      quadStream.write(quad)
+    }
+
+    quadStream.end()
+
+    if (!request) {
+      logger?.warn('No quads sent to the server. No request was made.')
+    }
+
+    await request
+  }
+}

packages/graph-store/lib/SinkToWritable.js

@@ -24,8 +24,6 @@
   "homepage": "https://github.com/zazuko/barnard59",
   "dependencies": {
     "duplex-to": "^1.0.0",
-    "onetime": "^6.0.0",
-    "promise-the-world": "^1.0.1",
     "readable-stream": "^3.6.0",
     "sparql-http-client": "^3.0.0",
     "barnard59-base": "^2.3.0",

packages/graph-store/lib/storeWritable.js

@@ -1,14 +1,16 @@
+import { Duplex } from 'node:stream'
 import Client from 'sparql-http-client'
-import SinkToWritable from './lib/SinkToWritable.js'
+import duplexTo from 'duplex-to'
 import { toTerm } from './lib/graph.js'
+import storeWritable from './lib/storeWritable.js'
 
 /**
  * @this {import('barnard59-core').Context}
  * @param {Pick<import('sparql-http-client/StreamClient.js').Options<any>, 'user' | 'password'> & {
  *   endpoint: string,
  *   graph: string | import('clownface').GraphPointer<import('@rdfjs/types').NamedNode> | import('@rdfjs/types').NamedNode,
  * }} options
- * @returns {import('readable-stream').Writable}
+ * @returns {import('node:stream').Writable}
  */
 function post({ endpoint, graph, user, password }) {
   if (!graph) {
@@ -22,9 +24,9 @@ function post({ endpoint, graph, user, password }) {
     password,
   })
 
-  return new SinkToWritable(readable => client.store.post(readable, {
+  return duplexTo.writable(Duplex.from(storeWritable(readable => client.store.post(readable, {
     graph: toTerm(this.env, graph),
-  }))
+  }), this)))
 }
 
 export default post

packages/graph-store/package.json

@@ -1,14 +1,16 @@
+import { Duplex } from 'node:stream'
 import Client from 'sparql-http-client'
-import SinkToWritable from './lib/SinkToWritable.js'
+import duplexTo from 'duplex-to'
 import { toTerm } from './lib/graph.js'
+import storeWritable from './lib/storeWritable.js'
 
 /**
  * @this {import('barnard59-core').Context}
  * @param {Pick<import('sparql-http-client/StreamClient.js').Options<any>, 'user' | 'password'> & {
  *   endpoint: string,
  *   graph: string | import('clownface').GraphPointer<import('@rdfjs/types').NamedNode> | import('@rdfjs/types').NamedNode,
  * }} options
- * @returns {import('readable-stream').Writable}
+ * @returns {import('node:stream').Writable}
  */
 function put({ endpoint, graph, user, password }) {
   if (!graph) {
@@ -22,9 +24,9 @@ function put({ endpoint, graph, user, password }) {
     password,
   })
 
-  return new SinkToWritable(readable => client.store.put(readable, {
+  return duplexTo.writable(Duplex.from(storeWritable(readable => client.store.put(readable, {
     graph: toTerm(this.env, graph),
-  }))
+  }), this)))
 }
 
 export default put

packages/graph-store/post.js

@@ -1,31 +1,17 @@
 import { strictEqual } from 'node:assert'
 import { promisify } from 'node:util'
+import { Readable, finished } from 'node:stream'
 import rdf from '@zazuko/env'
 import quadToNTriples from '@rdfjs/to-ntriples'
 import withServer from 'express-as-promise/withServer.js'
 import getStream from 'get-stream'
-import { isReadableStream as isReadable, isWritableStream as isWritable } from 'is-stream'
-import { finished } from 'readable-stream'
 import postUnbound from '../post.js'
 
 const post = postUnbound.bind({ env: rdf })
 
 const ns = rdf.namespace('http://example.org/')
 
 describe('post', () => {
-  it('should return a writable stream', async () => {
-    await withServer(async server => {
-      const baseUrl = await server.listen()
-
-      const stream = post({ endpoint: baseUrl, graph: 'default' })
-
-      strictEqual(isReadable(stream), false)
-      strictEqual(isWritable(stream), true)
-
-      stream.end()
-    })
-  })
-
   it('should send a POST request', async () => {
     await withServer(async server => {
       let called = false
@@ -40,10 +26,7 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, graph: ns.graph1 })
 
-      stream.write(quad)
-      stream.end()
-
-      await promisify(finished)(stream)
+      await getStream(Readable.from([quad]).pipe(stream))
 
       strictEqual(called, true)
     })
@@ -62,9 +45,7 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, graph: ns.graph1 })
 
-      stream.end()
-
-      await promisify(finished)(stream)
+      await getStream(Readable.from([]).pipe(stream))
 
       strictEqual(called, false)
     })
@@ -84,8 +65,7 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, graph: ns.graph1 })
 
-      stream.write(quad)
-      stream.end()
+      await getStream(Readable.from([quad]).pipe(stream))
 
       await promisify(finished)(stream)
 
@@ -115,13 +95,7 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, graph: ns.graph1 })
 
-      quads.forEach(quad => {
-        stream.write(quad)
-      })
-
-      stream.end()
-
-      await promisify(finished)(stream)
+      await getStream(Readable.from(quads).pipe(stream))
 
       strictEqual(content[quads[0].graph.value], expected)
     })
@@ -151,13 +125,7 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, graph: 'default' })
 
-      quads.forEach(quad => {
-        stream.write(quad)
-      })
-
-      stream.end()
-
-      await promisify(finished)(stream)
+      await getStream(Readable.from(quads).pipe(stream))
 
       strictEqual(typeof graph, 'undefined')
       strictEqual(content, expected)
@@ -178,8 +146,7 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, user: 'testuser', password: 'testpassword', graph: ns.graph1 })
 
-      stream.write(quad)
-      stream.end()
+      await getStream(Readable.from([quad]).pipe(stream))
 
       await promisify(finished)(stream)
 
@@ -198,13 +165,10 @@ describe('post', () => {
       const baseUrl = await server.listen()
       const stream = post({ endpoint: baseUrl, graph: ns.graph1 })
 
-      stream.write(quad)
-      stream.end()
-
       let error = null
 
       try {
-        await promisify(finished)(stream)
+        await getStream(Readable.from([quad]).pipe(stream))
       } catch (err) {
         error = err
       }