How do I implement a batch task in redux-saga with limited concurrency and sane cancellation?

Question

I'm trying to implement image uploads via redux-saga. The features I need to include are:

Concurrent uploads limit. This is achieved by using a channel as described in saga docs
The action I listen to, START_UPLOADS in the code below, contains a (possible quite long) array of files which are published to the channel separately.
I need to be able to cancel all current uploads via another action, CANCEL_ACTION, including those that arrived in any START_UPLOADS but weren't published to the channel yet, as well as those currently being processed in any of the uploadImage workers.

The code I arrived at is as follows. My problem with it is that the cancelAll handler is being executed AFTER the finally block that restarts the saga and the overall fact that I seem to need to restart everything. It looks clunky and error prone. Can you give any advice on whether this is how Sagas are meant to be used?

function* uploadImage(file) {
  const config = yield getConfig();
  const getRequest = new SagaRequest();
  console.log("Making async request here.");
}

function* consumeImages(uploadRequestsChannel) {
  while (true) {
    const fileAdded = yield take(uploadRequestsChannel);
    // process the request
    yield* uploadImage(fileAdded);
  }
}

function* uploadImagesSaga() {
  const CONCURRENT_UPLOADS = 10;
  const uploadRequestsChannel = yield call(channel);
  let workers = [];
  function* scheduleWorkers() {
    workers = [];
    for (let i = 0; i < CONCURRENT_UPLOADS; i++) {
      const worker = yield fork(consumeImages, uploadRequestsChannel);
      workers.push(worker);
    }
  }

  let listener;
  yield* scheduleWorkers();

  function* cancelAll() {
    // cancel producer and consumers, flush channel
    yield cancel(listener);
    for (const worker of workers) {
      yield cancel(worker);
    }
    yield flush(uploadRequestsChannel);
  }

  function* putToChannel(chan, task) {
    return yield put(chan, task);
  }

  function* listenToUploads() {
    try {
      while (true) {
        const { filesAdded } = yield take(START_UPLOADS);
        for (const fileAdded of filesAdded) {
          yield fork(putToChannel, uploadRequestsChannel, fileAdded);
        }
      }
    } finally {
      // if cancelled, restart consumers and producer
      yield* scheduleWorkers();
      listener = yield fork(listenToUploads);
    }
  }

  listener = yield fork(listenToUploads);

  while (true) {
    yield take(CANCEL_ACTION);
    yield call(cancelAll);
  }
}

export default uploadImagesSaga;

EDIT: distilled into a sandbox here: https://codesandbox.io/s/cancellable-counter-example-qomw6

user_582173 · Answer

I like using race for cancellation - the resolved value of the race is an object with one key and value (of the "winning" task). redux-saga race() docs

const result = yield race({
  cancel: take(CANCEL_ACTION),
  listener: call(listenToUploads), // use blocking `call`, not fork
});

if (result.cancel) {
  yield call(cancelAll)
}

^ This could be wrapped in a while (true) loop, so you should be able to consolidate the duplicated fork()s from the original example. If workers need to be re-scheduled, you could consider handling that inside cancelAll.

I prefer letting the outer-task handle restarts, rather than calling tasks from their own finally blocks.

Edit: refactored example sandbox https://codesandbox.io/s/cancellable-counter-example-j5vxr

How do I implement a batch task in redux-saga with limited concurrency and sane cancellation?

1 Answers