fix(map): handle errors that are thrown in res()

src/operators/Map.ts

@@ -4,6 +4,14 @@ import {FIO} from '../internals/FIO'
 import {REJ} from '../internals/REJ'
 import {RES} from '../internals/RES'
 
+const SafeResolve = <A>(a: A, rej: REJ, res: RES<A>) => {
+  try {
+    res(a)
+  } catch (e) {
+    rej(e as Error)
+  }
+}
+
 /**
  * @ignore
  */
@@ -14,6 +22,6 @@ export class Map<A, B> implements FIO<B> {
   ) {}
 
   public fork(sh: IScheduler, rej: REJ, res: RES<B>): Cancel {
-    return this.src.fork(sh, rej, a => res(this.ab(a)))
+    return this.src.fork(sh, rej, a => SafeResolve(this.ab(a), rej, res))
   }
 }

test/Map.test.ts

@@ -5,6 +5,8 @@ import {assert} from 'chai'
 
 import {IO} from '../'
 
+import {Counter} from './internals/Counter'
+import {ForkNRun} from './internals/ForkNRun'
 import {RejectingIOSpec, ResolvingIOSpec} from './internals/IOSpecification'
 
 describe('map', () => {
@@ -15,6 +17,18 @@ describe('map', () => {
     const expected = 11
     assert.equal(actual, expected)
   })
+  it('should capture exceptions on resolve', () => {
+    const counter = Counter()
+    const {timeline} = ForkNRun(
+      counter.inc.map(() => {
+        throw new Error('FAILURE')
+      })
+    )
+    const actual = timeline.list()
+    const expected = timeline.create(['REJECT', 1, 'Error: FAILURE'])
+
+    assert.deepStrictEqual(actual, expected)
+  })
   ResolvingIOSpec(() => IO.of(10).map(i => 100))
   RejectingIOSpec(() =>
     IO.of(10).map(i => {

test/internals/ForkNRun.ts

@@ -0,0 +1,19 @@
+/**
+ * Created by tushar on 2019-04-18
+ */
+
+import {FIO} from '../../src/internals/FIO'
+
+import {IOCollector} from './IOCollector'
+
+/**
+ * Helpful wrapper over IOCollector
+ * Forks the IO and runs everything in the queue.
+ */
+export const ForkNRun = <A>(io: FIO<A>) => {
+  const {fork, timeline, scheduler} = IOCollector(io)
+  fork()
+  scheduler.run()
+
+  return {timeline}
+}