deadlock when using `batch` with `flatMap` and mongo stream as source (node >= 10)

The following test deadlocks (`toCallback` never gets called):

```js
/* eslint-disable no-unused-vars, no-shadow, no-redeclare */
var _, EventEmitter = require('events').EventEmitter,
through = require('through'),
// sinon = require('sinon'),
Stream = require('stream'),
streamify = require('stream-array'),
concat = require('concat-stream'),
// Promise = RSVP.Promise,
transducers = require('transducers-js'),
bluebird = require('bluebird'),
runTask = require('orchestrator/lib/runTask'),
fl = require('fantasy-land'),
bufferFrom = require('buffer-from');

if (global.highland != null) {
_ = global.highland;
}
else {
_ = require('../lib/index');
}

// Use bluebird cancellation. We want to test against it.
bluebird.config({
cancellation: true,
longStackTraces : true
});

var MongoClient = require('mongodb').MongoClient

exports['batch - mongodb'] = function (test) {
test.expect(1);

MongoClient.connect('mongodb://localhost:27017')
.then(async (client) => {

var db = client.db("test")
var col = db.collection("test")
const docs = Array.from(Array(22).keys()).map( _id => { return { _id } })
await col.removeMany({})
await col.insertMany(docs)

cursor = col.find()
cursor.batchSize(2)
// Changing batch size to 3 causes the test to pass.
// cursor.batchSize(3)

_(cursor.stream())
.batch(11)
// Changing batch size to 12 causes the test to pass.
// .batch(12)
.flatMap( x => {
console.log("x:", x)
return _(x)
})
// If flatMap above is replaced with .map, the test passes.
// .map( x => {
// console.log("x:", x)
// return x
// })
.reduce(
(acc, x) => {
console.log("conc", acc, x)
return acc.concat(x)
},
[]
)
.toCallback((e, xs) => {
console.log("xs",xs)
test.same(xs, docs);
test.done();
})
})

};
```

You save save the above in `tests/batch_deadlock.js`, install mongo drives with `npm i mongodb` and finally run it with `npx nodeunit tests/batch_deadlock.js`. It should deadlock when running with node version >= 10, but passes when ran with node 8.x.

I've also left a few comments in the code which show some tweaks that make the deadlock go away.