Add cargo queue type and tests

[justinmchase]

I basically copied the code and tests from cargo. I then updated the function to expose the concurrency parameter and the comments to include copies from the queue code.

If there are docs in any other locations that need to be updated please let me know.

This fixes #1555

[aearly]

Thanks for doing this! A couple tweaks are needed though.

[aearly]

You don't need to redefine the CargoObject, you can refer to the one defined in cargo.js

[justinmchase]

Should I rename it here to be @typedef {Object} CargoQueueObject ?

[aearly]

The returned object has the same shape as the CargoObject, so there's no need for a duplicate JSDoc definition.

[aearly]

Again, you can refer to CargoObject here.

[aearly]

Looks like you left the .only() in!

[justinmchase]

whoops!

[aearly]

This test failed in one of our CI runs. Can you think of a way to re-write it so that it's more robust? We've been finding that using setTimeout isn't very consistent in CI.

[justinmchase]

Yeah I don't think setTimeout is actually deterministic, which can be a real pain for tests.

I thought about this for a while and it seems like the best solution would be a kind of elaborate recorder which would record the state of the queue at each worker invocation and callback. It would have to allow you to, instead of using time, use event frames. Using setImmediate is deterministic, the callbacks will be called in the order they were queued. Therefore your worker function and push callbacks could record the state of the queue and use setImmediate to move from frame to frame.

You could then have a way to describe which frames input is inserted and then have expectations set which are unordered. Every action would record the state of the queue in each frame.

This is total psuedo code but I was picturing something like this:

it('example', (done) => {
  var concurrency = 2
  var payload = 2
  var tester = new QueueTester(concurrency, payload) // contains cargoQueue
  tester.input([
    [1, 2],          // push at frame 0
    [3, 4, 5, 6]   // push at frame 1
  ])

  // For the above:
  // at frame 0 two items are enqueued
  // at frame 1 the first worker pops both items and processes them, 4 items are pushed
  // at frame 2 the first worker completes, the second worker pops 2 items and processes them
  // at frame 3 the first worker pops 2 items and processes them, the second worker completes
  // at frame 4 the first worker completes
  // at frame 5 the queue is idle and empty
  tester.expectation([
    { length: 2, workers: [] },
    {
      length: 0,
      workers: [
        { received: [1, 2] }
      ]
    },
    {
      length: 2,
      workers: [
        { completed: [1, 2] },
        { received: [3, 4] }
      ]
    },
    {
      length: 0,
      workers: [
        { received: [5, 6] },
        { completed: [3, 4] }
      ]
    },
    {
      length: 0,
      workers: [
        { completed: [5, 6]
      ]
    },
    {
      length: 0,
      workers: []
    }
  ])
})

tester.run(done)

I didn't go that far, instead I just altered the one failing test to use setImmedate instead of setTimeout and nested them so that they went in the same order as the original test. I think that's more deterministic but it sort of gross. Let me know if you had something else, more specific in mind.

[aearly]

Yeah, I've started using a similar strategy for newer tests, verifying concurrency at various checkpoints, and verifying the correct order of execution. It will be a pain to retrofit all the older setTimeout based tests though.

[aearly]

This is good enough for now, though.