Fixed throttle to accept a channel as originally intended (#2340)

* bug: throttle now accepts a channel as originally intended

Co-authored-by: Mateusz Burzyński <mateuszburzynski@gmail.com>
Closes: #1805

* Update .changeset/neat-wolves-wink.md

Co-authored-by: Mateusz Burzyński <mateuszburzynski@gmail.com>

.changeset/neat-wolves-wink.md

@@ -0,0 +1,6 @@
+---
+'redux-saga': patch
+'@redux-saga/core': patch
+---
+
+throttle now accepts a channel as originally intended

packages/core/__tests__/sagaHelpers/throttle.js

@@ -1,7 +1,8 @@
-import sagaMiddleware, { END } from '../../src'
+import sagaMiddleware, { buffers, END } from '../../src'
 import { createStore, applyMiddleware } from 'redux'
 import delayP from '@redux-saga/delay-p'
-import { take, cancel, throttle } from '../../src/effects'
+import { take, cancel, throttle, fork } from '../../src/effects'
+import { channel } from '../../src'
 
 test('throttle', () => {
   jest.useFakeTimers()
@@ -54,6 +55,70 @@ test('throttle', () => {
   )
 })
 
+test('throttle - channel', () => {
+  jest.useFakeTimers()
+
+  const actual = []
+  const expected = [['a1', 'a2', 0], ['a1', 'a2', 10], ['a1', 'a2', 20], ['a1', 'a2', 30], ['a1', 'a2', 34]]
+  const middleware = sagaMiddleware()
+  const store = applyMiddleware(middleware)(createStore)(() => {})
+  middleware.run(root)
+
+  function* root() {
+    const chan = channel(buffers.sliding(1))
+    yield fork(listen, chan)
+    yield fork(worker1, chan)
+  }
+
+  function* listen(chan) {
+    while (true) {
+      const action = yield take('ACTION')
+      chan.put(action)
+    }
+  }
+
+  function* worker1(chan) {
+    const task = yield throttle(100, chan, worker2, 'a1', 'a2')
+    yield take('CANCEL_WATCHER')
+    yield cancel(task)
+  }
+
+  function* worker2(arg1, arg2, { payload }) {
+    actual.push([arg1, arg2, payload])
+  }
+
+  const dispatchedActions = []
+
+  for (let i = 0; i < 35; i++) {
+    dispatchedActions.push(
+      delayP(i * 10)
+        .then(() => store.dispatch({ type: 'ACTION', payload: i }))
+        .then(() => jest.advanceTimersByTime(10)), // next tick
+    )
+  }
+
+  Promise.resolve()
+    .then(() => jest.advanceTimersByTime(1)) // just start for the smallest tick
+    .then(() => jest.advanceTimersByTime(10)) // tick past first delay
+
+  return (
+    dispatchedActions[34] // wait so trailing dispatch gets processed
+      .then(() => jest.advanceTimersByTime(100))
+      .then(() => store.dispatch({ type: 'CANCEL_WATCHER' }))
+      // shouldn't be processed cause of getting canceled
+      .then(() => store.dispatch({ type: 'ACTION', payload: 40 }))
+      .then(() => {
+        // throttle must ignore incoming actions during throttling interval
+        expect(actual).toEqual(expected)
+        jest.useRealTimers()
+      })
+      .catch(err => {
+        jest.useRealTimers()
+        throw err
+      })
+  )
+})
+
 test('throttle: pattern END', () => {
   const delayMs = 20
   const middleware = sagaMiddleware()

packages/core/src/internal/io-helpers.js

@@ -39,13 +39,13 @@ export function takeLeading(patternOrChannel, worker, ...args) {
   return fork(takeLeadingHelper, patternOrChannel, worker, ...args)
 }
 
-export function throttle(ms, pattern, worker, ...args) {
+export function throttle(ms, patternOrChannel, worker, ...args) {
   if (process.env.NODE_ENV !== 'production') {
-    check(pattern, is.notUndef, 'throttle requires a pattern')
+    check(patternOrChannel, is.notUndef, `throttle requires a pattern or channel`)
     check(worker, is.notUndef, 'throttle requires a saga parameter')
   }
 
-  return fork(throttleHelper, ms, pattern, worker, ...args)
+  return fork(throttleHelper, ms, patternOrChannel, worker, ...args)
 }
 
 export function retry(maxTries, delayLength, worker, ...args) {

packages/core/src/internal/sagaHelpers/throttle.js

@@ -1,21 +1,28 @@
+import * as is from '@redux-saga/is'
 import fsmIterator, { safeName } from './fsmIterator'
 import { take, fork, actionChannel, delay } from '../io'
 import * as buffers from '../buffers'
 
-export default function throttle(delayLength, pattern, worker, ...args) {
+export default function throttle(delayLength, patternOrChannel, worker, ...args) {
   let action, channel
 
-  const yActionChannel = { done: false, value: actionChannel(pattern, buffers.sliding(1)) }
   const yTake = () => ({ done: false, value: take(channel) })
   const yFork = ac => ({ done: false, value: fork(worker, ...args, ac) })
   const yDelay = { done: false, value: delay(delayLength) }
 
   const setAction = ac => (action = ac)
   const setChannel = ch => (channel = ch)
 
+  const needsChannel = !is.channel(patternOrChannel)
+
+  if (!needsChannel) {
+    setChannel(patternOrChannel)
+  }
+
   return fsmIterator(
     {
       q1() {
+        const yActionChannel = { done: false, value: actionChannel(patternOrChannel, buffers.sliding(1)) }
         return { nextState: 'q2', effect: yActionChannel, stateUpdater: setChannel }
       },
       q2() {
@@ -28,7 +35,7 @@ export default function throttle(delayLength, pattern, worker, ...args) {
         return { nextState: 'q2', effect: yDelay }
       },
     },
-    'q1',
-    `throttle(${safeName(pattern)}, ${worker.name})`,
+    needsChannel ? 'q1' : 'q2',
+    `throttle(${safeName(patternOrChannel)}, ${worker.name})`,
   )
 }