test(subscribe, iterators): Fix error handling.

Author: timoxley

src/subscribe/index.ts

@@ -10,9 +10,15 @@ import MessagePipeline from './pipeline'
 import Validator from './Validator'
 import messageStream from './messageStream'
 import resendStream from './resendStream'
-import { MaybeAsync, Todo } from '../types'
+import { Todo, MaybeAsync } from '../types'
 import StreamrClient, { StreamPartDefinition, SubscribeOptions } from '..'
 
+async function defaultOnFinally(err?: Error) {
+    if (err) {
+        throw err
+    }
+}
+
 /**
  * @category Important
  */
@@ -39,7 +45,7 @@ export class Subscription extends Emitter {
     /** @internal */
     iterated?: Todo
 
-    constructor(client: StreamrClient, opts: Todo, onFinally = async () => {}) {
+    constructor(client: StreamrClient, opts: Todo, onFinally = defaultOnFinally) {
         super()
         this.client = client
         this.options = validateOptions(opts)
@@ -400,7 +406,7 @@ class Subscriptions {
         this.subSessions = new Map()
     }
 
-    async add(opts: StreamPartDefinition, onFinally: MaybeAsync<(err?: any) => void> = async () => {}) {
+    async add(opts: StreamPartDefinition, onFinally: MaybeAsync<(err?: any) => void> = defaultOnFinally) {
         const options = validateOptions(opts)
         const { key } = options
 

src/subscribe/pipeline.js

@@ -22,7 +22,7 @@ async function collect(src) {
  * Subscription message processing pipeline
  */
 
-export default function MessagePipeline(client, opts = {}, onFinally = async () => {}) {
+export default function MessagePipeline(client, opts = {}, onFinally = async (err) => { if (err) { throw err } }) {
     const options = validateOptions(opts)
     const { key, afterSteps = [], beforeSteps = [] } = options
     const id = counterId('MessagePipeline') + key
@@ -128,7 +128,7 @@ export default function MessagePipeline(client, opts = {}, onFinally = async ()
         // custom pipeline steps
         ...afterSteps
     ], async (err, ...args) => {
-        // await msgStream.cancel(err)
+        await msgStream.cancel(err)
         try {
             if (err) {
                 await onError(err)

src/utils/iterators.js

@@ -351,14 +351,17 @@ export function pipeline(iterables = [], onFinally = defaultOnFinally, { end, ..
         firstSrc = v
     }
 
-    const last = iterables.reduce((_prev, next, index) => {
-        let prev
-        let nextIterable
+    iterables.forEach((nextIterable) => {
+        if (nextIterable.cancel) {
+            cancelFns.add(nextIterable)
+        }
+    })
 
+    const last = iterables.reduce((_prev, next, index) => {
         const it = CancelableGenerator((async function* Gen() {
-            prev = index === 0 ? firstSrc : _prev
+            const prev = index === 0 ? firstSrc : _prev
             // take first "prev" from outer iterator, if one exists
-            nextIterable = typeof next === 'function' ? next(prev) : next
+            const nextIterable = typeof next === 'function' ? next(prev) : next
 
             if (prev && nextIterable[isPipeline]) {
                 nextIterable.setFirstSource(prev)
@@ -372,7 +375,6 @@ export function pipeline(iterables = [], onFinally = defaultOnFinally, { end, ..
                 prev.id = prev.id || 'inter-' + nextIterable.id
                 nextIterable.from(prev, { end })
             }
-
             yield* nextIterable
         }()), async (err) => {
             if (!error && err && error !== err) {
@@ -384,7 +386,8 @@ export function pipeline(iterables = [], onFinally = defaultOnFinally, { end, ..
         return it
     }, undefined)
 
-    pipelineValue = iteratorFinally(last, async () => {
+    pipelineValue = iteratorFinally(last, async (err) => {
+        error = AggregatedError.from(error, err)
         if (!cancelled) {
             await cancelAll(error)
         }