experimental success-errors wrapper. Fails when a reduce is used in t…

…he errors pipeline

test/async-pipeline-errors.test.js

@@ -0,0 +1,65 @@
+const _ = require('../src/index.js')
+const { pipeline } = _
+
+const successErrorsFlow = require('./success-errors-flow')
+
+test('pipeline with simple map is called', async () => {
+  const sourceFlow = _([1, 2]).map(x => {
+    if (x > 1) throw new Error(`${x} is wrong`)
+    return x * 10
+  })
+  const eachError = jest.fn()
+  const print = jest.fn()
+  const field = 'message'
+  const title = 'Errors:'
+  const errorsPipeline = pipeline()
+    .map(x => ({ ...x, just: 'mapped' }))
+    .tap(print)
+
+  const wrap = successErrorsFlow({
+    eachError,
+    errorsPipeline,
+  })
+  const stream = wrap(sourceFlow)
+  const results = await stream.toPromise()
+  expect(results).toEqual([10])
+  expect(eachError).toHaveBeenCalledTimes(1)
+  expect(print).toHaveBeenCalledTimes(1)
+  // expect(eachError).toHaveBeenCalledWith('')
+  // expect(print).toHaveBeenCalledWith('')
+})
+
+test('pipeline is not called', async () => {
+  const sourceFlow = _([1, 2]).map(x => {
+    if (x > 1) throw new Error(`${x} is too high`)
+    return x * 10
+  })
+  const eachError = jest.fn()
+  const print = jest.fn()
+  const field = 'message'
+  const title = 'Errors:'
+  const errorsPipeline = pipeline()
+    .reduce((aggregate, item) => {
+      const { total, messages } = aggregate
+      const { [field]: message } = item
+      console.log(title, { message })
+      return {
+        total: total + 1,
+        messages: [...messages, message],
+      }
+    }, { messages: [], total: 0 })
+    .tap(console.log)
+    .tap(print)
+
+  const wrap = successErrorsFlow({
+    eachError,
+    errorsPipeline,
+  })
+  const stream = wrap(sourceFlow)
+  const results = await stream.toPromise()
+  expect(results).toEqual([10])
+  expect(eachError).toHaveBeenCalledTimes(1)
+  expect(print).toHaveBeenCalledTimes(1)
+  // expect(eachError).toHaveBeenCalledWith('')
+  // expect(print).toHaveBeenCalledWith('')
+})

test/error-to-json.js

@@ -0,0 +1,10 @@
+module.exports = error => ({
+  type: 'error',
+  input: error.exstreamInput,
+  message: error.message,
+  stack: error.stack,
+  cause: error.cause
+    ? { message: error.cause.message, stack: error.cause.stack }
+    : 'Not Specified',
+  // error,
+})

test/success-errors-flow.js

@@ -0,0 +1,32 @@
+const _ = require('../src/index.js')
+const { pipeline, curry } = _
+const errorToJson = require('./error-to-json')
+
+// eslint-disable-next-line no-console
+const pattern = (
+  {
+    logger = console,
+    eachError = error => logger.warn(error),
+    stopOnError = false,
+    errorsPipeline = pipeline(),
+  },
+  sourceFlow) => {
+  const isWrappable = Array.isArray(sourceFlow) || sourceFlow.__exstream__ === true
+  const success = isWrappable
+    ? _(sourceFlow)
+    : sourceFlow()
+  const errors = _()
+  errors
+    .through(errorsPipeline)
+    .stopOnError(error => logger.error(error))
+    .each(eachError)
+  success.on('end', () => errors.end())
+
+  return stopOnError === true
+    ? success
+      .stopOnError(error => errors.write(errorToJson(error)))
+    : success
+      .errors(error => errors.write(errorToJson(error)))
+}
+
+module.exports = curry(pattern)