So returning stuff in the callback or by doing this.push() doesn't work. Major issue.

I ran into a pretty insidious bug that has taken me days to unravel, and I wanted to share it for anyone else who runs into the same problem.

The crux of the issue is that through2-concurrent requires that it's callback function (the one you the user provides) always be asynchronous. Clearly this is not strictly required by the library, but If that callback is not always asynchronous, it's possible for the event loop to block.

Let me share my code:

  return throughConcurrent.obj({ maxConcurrency: concurrency }, async function(
    domain,
    enc,
    done
  ) {
    try {
      const url = await frontierList.getNextUrlForDomain(domain);
      if (url) {
        this.push(url);
      }
    } catch (err) {
      console.error(err)
    }
    done();
  });
};

What happens is that when the done() function is invoked inside my callback, my callback is synchronously invoked with a domain. This is all fine, except If that domain also causes frontierList.getNextUrlForDomain(domain) to return immediately without performing an async operation, invoking done causes my callback is once again be invoked synchronously. What happens, then, is that sometimes I get into a pattern where maybe some percentage of async callbacks never finish because every subsequent value causes frontierList.getNextUrlForDomain(domain) to return immediately without performing an async operation. The event loop is quickly blocked, and my async operations never finish.

A workaround for this was pretty simple. I was already using a wrapping around through-concurrent, but I will only share the relevant parts here.

function createThroughConcurrent(concurrency, callback) {
  const stream = throughConcurrent.obj({ maxConcurrency: concurrency }, function(
    domain,
    enc,
    done
  ) {
    setImmediate(() => {
      callback.call(this, domain, enc, done);
    });
  });
  return stream;
};

The idea is that we want to clear the event loop no matter what before invoking our callback. This code accomplishes that.

If one is using through2-concurrent across multiple modules at the same time in a node application, which each module defining its own level of maximum concurrency, the last one in wins. That is, all the instances will use that level of max concurrency. This is because the maxConcurrency variable is defined without the 'var' keyword causing it to be defined globally.

No callback() statement. Example hangs after reaching limit.

When use it to implement a writeable stream, the 'finish' event fired before all of the pending jobs done.

As an option, not appropriate for a lot of use cases.

It would need to have a limit on the output buffer (the list of chunks that have arrived out of order and are waiting for the the one in front of them to be ready) otherwise it would be easy for it to use up all available memory if a particular chunk took a long time to process. This probably wouldn't be the same as the maxConcurrency option, this should continue to mean how many things can be being processed at a given time.

Hi there, great project!

Would it be possible to add git tags? This would automatically create GitHub releases. GitHub releases makes it easy to follow new releases with a RSS reader.

Projects like release-it automate this process.