Merge branch 'feature/tap'

CHANGELOG.md

@@ -1,3 +1,7 @@
+## Unreleased
+
+* Add `Signal#tap` method
+
 ## 6.0.0 (2019-03-23)
 
 * Refactor combinators to fix cyclical dependency warnings

packages/bulb/src/Signal.js

@@ -31,6 +31,7 @@ import switchMap from './combinators/switchMap'
 import take from './combinators/take'
 import takeUntil from './combinators/takeUntil'
 import takeWhile from './combinators/takeWhile'
+import tap from './combinators/tap'
 import throttle from './combinators/throttle'
 import zipWith from './combinators/zipWith'
 import { asap } from './scheduler'
@@ -1276,6 +1277,28 @@ export default class Signal {
     return new Signal(takeWhile(p, this))
   }
 
+  /**
+   * Performs the side effect function `f` for each value emitted by the
+   * signal. The returned signal contains the same events as the original
+   * signal.
+   *
+   * @param {Function} f The function to apply to each value emitted by the
+   * signal.
+   * @returns {Signal} A new signal.
+   * @example
+   *
+   * import { Signal } from 'bulb'
+   *
+   * const s = Signal
+   *   .of(1, 2, 3)
+   *   .tap(console.log)
+   *
+   * s.subscribe() // 1, 2, 3
+   */
+  tap (f) {
+    return new Signal(tap(f, this))
+  }
+
   /**
    * Limits the rate at which values are emitted by the signal to one every `n`
    * milliseconds. Values will be dropped when the rate limit is exceeded.

packages/bulb/src/Signal.test.js

@@ -30,6 +30,7 @@ import switchMap from './combinators/switchMap'
 import take from './combinators/take'
 import takeUntil from './combinators/takeUntil'
 import takeWhile from './combinators/takeWhile'
+import tap from './combinators/tap'
 import throttle from './combinators/throttle'
 import zipWith from './combinators/zipWith'
 import { asap } from './scheduler'
@@ -59,6 +60,7 @@ jest.mock('./combinators/switchMap', () => jest.fn(() => () => {}))
 jest.mock('./combinators/take', () => jest.fn(() => () => {}))
 jest.mock('./combinators/takeUntil', () => jest.fn(() => () => {}))
 jest.mock('./combinators/takeWhile', () => jest.fn(() => () => {}))
+jest.mock('./combinators/tap', () => jest.fn(() => () => {}))
 jest.mock('./combinators/throttle', () => jest.fn(() => () => {}))
 jest.mock('./combinators/zipWith', () => jest.fn(() => () => {}))
 jest.mock('./scheduler')
@@ -727,6 +729,17 @@ describe('Signal', () => {
     })
   })
 
+  describe('#tap', () => {
+    it('calls the combinator', () => {
+      const f = jest.fn()
+      const s = mockSignal()
+
+      s.tap(f)
+
+      expect(tap).toHaveBeenLastCalledWith(f, s)
+    })
+  })
+
   describe('#throttle', () => {
     it('calls the combinator', () => {
       const s = mockSignal()

packages/bulb/src/combinators/tap.js

@@ -0,0 +1,18 @@
+/**
+ * Performs the side effect function `f` for each value emitted by the signal
+ * `s`.
+ *
+ * @private
+ */
+export default function tap (f, s) {
+  return emit => {
+    const subscription = s.subscribe({ ...emit,
+      next (a) {
+        f(a)
+        emit.next(a)
+      }
+    })
+
+    return () => subscription.unsubscribe()
+  }
+}

packages/bulb/src/combinators/tap.test.js

@@ -0,0 +1,57 @@
+import { id } from 'fkit'
+
+import tap from './tap'
+import mockSignal from '../internal/mockSignal'
+
+let s
+let next, error, complete
+let emit
+
+describe('tap', () => {
+  beforeEach(() => {
+    s = mockSignal()
+
+    next = jest.fn()
+    error = jest.fn()
+    complete = jest.fn()
+
+    emit = { next, error, complete }
+  })
+
+  it('performs a side effect', () => {
+    const f = jest.fn()
+
+    tap(f, s)(emit)
+
+    expect(next).not.toHaveBeenCalled()
+    s.next(1)
+    expect(f).toHaveBeenLastCalledWith(1)
+    expect(next).toHaveBeenCalledTimes(1)
+    expect(next).toHaveBeenLastCalledWith(1)
+  })
+
+  it('emits an error when the given signal emits an error', () => {
+    tap(id, s)(emit)
+
+    expect(error).not.toHaveBeenCalled()
+    s.error('foo')
+    expect(error).toHaveBeenCalledTimes(1)
+    expect(error).toHaveBeenCalledWith('foo')
+  })
+
+  it('completes when the given signal is completed', () => {
+    tap(id, s)(emit)
+
+    expect(complete).not.toHaveBeenCalled()
+    s.complete()
+    expect(complete).toHaveBeenCalledTimes(1)
+  })
+
+  it('unmounts the given signal when the unsubscribe function is called', () => {
+    const unsubscribe = tap(id, s)(emit)
+
+    expect(s.unmount).not.toHaveBeenCalled()
+    unsubscribe()
+    expect(s.unmount).toHaveBeenCalledTimes(1)
+  })
+})