feat(map,filter): additional arguments

.changeset/selfish-cougars-carry.md

@@ -0,0 +1,5 @@
+---
+"barnard59-base": minor
+---
+
+Adds a variadic parameter to `map` and` filter` steps, which will be forwarded to their respecitve callback functions called for each chunk

packages/base/filter.js

@@ -1,20 +1,23 @@
 import { obj } from 'through2'
 
 /**
- * @typedef {(this: import('barnard59-core').Context, chunk: T, encoding: string) => boolean | Promise<boolean>} Filter<T>
+ * @typedef {(this: import('barnard59-core').Context, chunk: T, encoding: string, ...args: A) => boolean | Promise<boolean>} Filter<T, A>
  * @template T
+ * @template {Array<unknown>} A
  */
 
 /**
  * @template T
+ * @template {Array<unknown>} A
  * @this {import('barnard59-core').Context}
- * @param {Filter<T>} func
+ * @param {Filter<T, A>} func
+ * @param {A} args
  * @return {import('stream').Transform}
  */
-function filter(func) {
+function filter(func, ...args) {
   return obj((chunk, encoding, callback) => {
     Promise.resolve().then(() => {
-      return func.call(this, chunk, encoding)
+      return func.call(this, chunk, encoding, ...args)
     }).then(result => {
       if (result) {
         return callback(null, chunk)

packages/base/map.js

@@ -1,31 +1,35 @@
 import transform from 'parallel-transform'
 
 /**
- * @typedef {(this: import('barnard59-core').Context, chunk: From) => Promise<To> | To} MapCallback
+ * @typedef {(this: import('barnard59-core').Context, chunk: From, ...args: Args) => Promise<To> | To} MapCallback
  * @template From, To
+ * @template {Array<unknown>} Args
  */
 
 /**
  * @typedef {{
- *   map: MapCallback<From, To>
+ *   map: MapCallback<From, To, Args>
  *   concurrency?: number
  *   ordered?: boolean
  *   objectMode?: boolean
- * }|MapCallback<From, To>} MapOptions
+ * }|MapCallback<From, To, Args>} MapOptions
  * @template From, To
+ * @template {Array<unknown>} Args
  */
 
 /**
  * Processes chunks with a transform function
  *
  * @this {import('barnard59-core').Context}
- * @param {MapOptions<From, To>} options Transform function or complex options
+ * @param {MapOptions<From, To, Args>} options Transform function or complex options
+ * @param {Args} args Additional arguments to pass to the transform function
  * @return {import('stream').Transform}
  * @template From, To
+ * @template {Array<unknown>} Args
  */
-export default function map(options) {
+export default function map(options, ...args) {
   /**
-   * @type {MapCallback<*, *>}
+   * @type {MapCallback<*, *, Args>}
    */
   let func
   let concurrency = 1
@@ -42,7 +46,7 @@ export default function map(options) {
 
   return transform(concurrency, { ordered, objectMode }, (data, callback) => {
     Promise.resolve().then(() => {
-      return func.call(this, data)
+      return func.call(this, data, ...args)
     }).then(result => {
       callback(null, result)
     }).catch(callback)

packages/base/test/filter.test.js

@@ -5,6 +5,7 @@ import filter from '../filter.js'
 
 describe('filter', () => {
   it('should pass pipeline context to callback function', async () => {
+    // given
     const input = new Readable({
       objectMode: true,
       read: () => {
@@ -17,11 +18,34 @@ describe('filter', () => {
       variable: new Map().set('condition', 'a'),
     }
 
+    // when
     const outStream = input.pipe(filter.call(context, function (chunk) {
       return this.variable.get('condition') === chunk
     }))
     const output = await array(outStream)
 
+    // then
     deepStrictEqual(output, ['a'])
   })
+
+  it('should pass additional arguments to callback function', async () => {
+    // given
+    const input = new Readable({
+      objectMode: true,
+      read: () => {
+        input.push(1)
+        input.push(2)
+        input.push(3)
+        input.push(null)
+      },
+    })
+
+    // when
+    const greaterThan = (chunk, _, minValue) => chunk > minValue
+    const outStream = input.pipe(filter.call(context, greaterThan, 2))
+    const output = await array(outStream)
+
+    // then
+    deepStrictEqual(output, [3])
+  })
 })

packages/base/test/map.test.js

@@ -25,6 +25,40 @@ describe('map', () => {
     deepStrictEqual(output, ['foo_a', 'foo_b'])
   })
 
+  it('should pass optional arguments to mapper function', async () => {
+    const input = new Readable({
+      read: () => {
+        input.push('a')
+        input.push('b')
+        input.push(null)
+      },
+    })
+
+    const prependPrefix = (chunk, prefix) => prefix + chunk
+    const outStream = input.pipe(map.call(context, prependPrefix, 'foo_'))
+    const output = await array(outStream)
+
+    deepStrictEqual(output, ['foo_a', 'foo_b'])
+  })
+
+  it('should pass optional arguments to mapper function when used with options object', async () => {
+    const input = new Readable({
+      read: () => {
+        input.push('a')
+        input.push('b')
+        input.push(null)
+      },
+    })
+
+    const prependPrefix = (chunk, prefix) => prefix + chunk
+    const outStream = input.pipe(map.call(context, {
+      map: prependPrefix,
+    }, 'foo_'))
+    const output = await array(outStream)
+
+    deepStrictEqual(output, ['foo_a', 'foo_b'])
+  })
+
   it('accepts a function as parameter', async () => {
     // given
     const transform = letter => letter.toUpperCase()

test/e2e/definitions/foreach/with-variable.ttl

@@ -36,6 +36,12 @@
     [
       p:stepList
         (
-          [ base:map ( "filename => this.variables.get('root') + filename"^^code:EcmaScript ) ]
+          [
+            base:map
+              (
+                "(filename, root) => root + filename"^^code:EcmaScript
+                "root"^^p:VariableName
+              )
+          ]
         )
     ] .

test/e2e/forEach.e2e.test.js

@@ -1,8 +1,9 @@
-import { deepStrictEqual, strictEqual } from 'node:assert'
+import { strictEqual } from 'node:assert'
 import { createPipeline } from 'barnard59-core'
 import getStream from 'get-stream'
 import { pipelineDefinitionLoader } from 'barnard59-test-support/loadPipelineDefinition.js'
 import env from 'barnard59-env'
+import { expect } from 'chai'
 
 const loadPipelineDefinition = pipelineDefinitionLoader(import.meta.url, 'definitions')
 
@@ -36,7 +37,7 @@ describe('forEach', () => {
 
     const out = await getStream.array(pipeline.stream)
 
-    deepStrictEqual(out, [
+    expect(out).to.contain.all.members([
       '/root/definitions/foreach/csv-duplicate.ttl',
       '/root/definitions/foreach/with-handler.ttl',
       '/root/definitions/foreach/with-variable.ttl',