Skip to content

Commit accdde5

Browse files
RubenVerborghRubenVerborgh
committedJul 4, 2020
Allow customizing the task scheduler.
Closes #17
·
v3.9.0v3.0.0-beta.6
1 parent 4e85c70 commit accdde5

8 files changed

+109
-41
lines changed
 

‎asynciterator.ts

+30-11
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,23 @@
66
import { EventEmitter } from 'events';
77
import queueMicrotask from 'queue-microtask';
88

9+
let taskScheduler: TaskScheduler = queueMicrotask;
10+
11+
/** Schedules the given ask for asynchronous execution. */
12+
export function scheduleTask(task: () => void): void {
13+
taskScheduler(task);
14+
}
15+
16+
/** Returns the asynchronous task scheduler. */
17+
export function getTaskScheduler(): TaskScheduler {
18+
return taskScheduler;
19+
}
20+
21+
/** Sets the asynchronous task scheduler. */
22+
export function setTaskScheduler(scheduler: TaskScheduler): void {
23+
taskScheduler = scheduler;
24+
}
25+
926
/**
1027
ID of the INIT state.
1128
An iterator is initializing if it is preparing main item generation.
@@ -90,7 +107,7 @@ export class AsyncIterator<T> extends EventEmitter {
90107
if (!eventAsync)
91108
this.emit('end');
92109
else
93-
queueMicrotask(() => this.emit('end'));
110+
taskScheduler(() => this.emit('end'));
94111
}
95112
}
96113
return valid;
@@ -210,7 +227,7 @@ export class AsyncIterator<T> extends EventEmitter {
210227
@protected
211228
*/
212229
protected _endAsync() {
213-
queueMicrotask(() => this._end());
230+
taskScheduler(() => this._end());
214231
}
215232

216233
/**
@@ -236,7 +253,7 @@ export class AsyncIterator<T> extends EventEmitter {
236253
this._readable = readable;
237254
// If the iterator became readable, emit the `readable` event
238255
if (readable)
239-
queueMicrotask(() => this.emit('readable'));
256+
taskScheduler(() => this.emit('readable'));
240257
}
241258
}
242259

@@ -308,7 +325,7 @@ export class AsyncIterator<T> extends EventEmitter {
308325
return properties && properties[propertyName];
309326
// If the value has been set, send it through the callback
310327
if (properties && (propertyName in properties)) {
311-
queueMicrotask(() => callback(properties[propertyName]));
328+
taskScheduler(() => callback(properties[propertyName]));
312329
}
313330
// If the value was not set, store the callback for when the value will be set
314331
else {
@@ -336,7 +353,7 @@ export class AsyncIterator<T> extends EventEmitter {
336353
const callbacks = propertyCallbacks[propertyName];
337354
if (callbacks) {
338355
delete propertyCallbacks[propertyName];
339-
queueMicrotask(() => {
356+
taskScheduler(() => {
340357
for (const callback of callbacks)
341358
callback(value);
342359
});
@@ -502,7 +519,7 @@ function waitForDataListener(this: AsyncIterator<any>, eventName: string) {
502519
this.removeListener('newListener', waitForDataListener);
503520
addSingleListener(this, 'readable', emitData);
504521
if (this.readable)
505-
queueMicrotask(() => emitData.call(this));
522+
taskScheduler(() => emitData.call(this));
506523
}
507524
}
508525
// Emits new items though `data` events as long as there are `data` listeners
@@ -710,7 +727,7 @@ export class BufferedIterator<T> extends AsyncIterator<T> {
710727
constructor({ maxBufferSize = 4, autoStart = true } = {}) {
711728
super(INIT);
712729
this.maxBufferSize = maxBufferSize;
713-
queueMicrotask(() => this._init(autoStart));
730+
taskScheduler(() => this._init(autoStart));
714731
}
715732

716733
/**
@@ -886,7 +903,7 @@ export class BufferedIterator<T> extends AsyncIterator<T> {
886903
// Acquire reading lock to avoid recursive reads
887904
if (!this._reading) {
888905
this._reading = true;
889-
queueMicrotask(() => {
906+
taskScheduler(() => {
890907
// Release reading lock so _fillBuffer` can take it
891908
this._reading = false;
892909
this._fillBuffer();
@@ -1072,7 +1089,7 @@ export class TransformIterator<S, D = S> extends BufferedIterator<D> {
10721089
const next = () => {
10731090
// Continue transforming until at least `count` items have been pushed
10741091
if (this._pushedCount < count && !this.closed)
1075-
queueMicrotask(() => this._readAndTransform(next, done));
1092+
taskScheduler(() => this._readAndTransform(next, done));
10761093
else
10771094
done();
10781095
};
@@ -1220,10 +1237,10 @@ export class SimpleTransformIterator<S, D = S> extends TransformIterator<S, D> {
12201237
/* Tries to read and transform items */
12211238
protected _read(count: number, done: () => void) {
12221239
const next = () => this._readAndTransformSimple(count, nextAsync, done);
1240+
this._readAndTransformSimple(count, nextAsync, done);
12231241
function nextAsync() {
1224-
queueMicrotask(next);
1242+
taskScheduler(next);
12251243
}
1226-
this._readAndTransformSimple(count, nextAsync, done);
12271244
}
12281245

12291246
/* Reads and transform items */
@@ -1858,3 +1875,5 @@ type SourceExpression<T> =
18581875

18591876
type InternalSource<T> =
18601877
AsyncIterator<T> & { _destination: AsyncIterator<any> };
1878+
1879+
type TaskScheduler = (task: () => void) => void;

‎test/AsyncIterator-test.js

+2-2
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,10 @@ import {
44
CLOSED,
55
ENDED,
66
DESTROYED,
7+
scheduleTask,
78
} from '../asynciterator.mjs';
89

910
import { EventEmitter } from 'events';
10-
import queueMicrotask from 'queue-microtask';
1111

1212
describe('AsyncIterator', () => {
1313
describe('The AsyncIterator module', () => {
@@ -408,7 +408,7 @@ describe('AsyncIterator', () => {
408408
before(() => {
409409
iterator = new AsyncIterator();
410410
captureEvents(iterator, 'data', 'readable', 'end');
411-
iterator._destroy = (error, callback) => queueMicrotask(callback);
411+
iterator._destroy = (error, callback) => scheduleTask(callback);
412412
iterator.destroy();
413413
});
414414

‎test/BufferedIterator-test.js

+9-11
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,10 @@ import {
22
AsyncIterator,
33
BufferedIterator,
44
CLOSED,
5+
scheduleTask,
56
} from '../asynciterator.mjs';
67

78
import { EventEmitter } from 'events';
8-
import queueMicrotask from 'queue-microtask';
99

1010
describe('BufferedIterator', () => {
1111
describe('The BufferedIterator function', () => {
@@ -301,7 +301,7 @@ describe('BufferedIterator', () => {
301301
function createIterator(options) {
302302
const iterator = new BufferedIterator(options);
303303
iterator._read = function (count, done) {
304-
queueMicrotask(() => {
304+
scheduleTask(() => {
305305
this.close();
306306
done();
307307
});
@@ -660,8 +660,8 @@ describe('BufferedIterator', () => {
660660
describe('after `read` is called and the iterator has been closed', () => {
661661
before(() => {
662662
iterator.read();
663-
queueMicrotask(() => { _readDone(); });
664-
queueMicrotask(() => { iterator.close(); });
663+
scheduleTask(() => { _readDone(); });
664+
scheduleTask(() => { iterator.close(); });
665665
});
666666

667667
it('should have emitted the `end` event', () => {
@@ -1292,7 +1292,7 @@ describe('BufferedIterator', () => {
12921292
const iterator = new BufferedIterator(options);
12931293
iterator._read = function (count, done) {
12941294
this._push('a');
1295-
queueMicrotask(() => {
1295+
scheduleTask(() => {
12961296
this._push('b');
12971297
this._push('c');
12981298
done();
@@ -1408,9 +1408,7 @@ describe('BufferedIterator', () => {
14081408
before(done => {
14091409
iterator = new BufferedIterator({ autoStart: false });
14101410
iterator._read = (count, callback) => { readDone = callback; };
1411-
// `queueMicrotask` because reading directly after construction does not call `_read`;
1412-
// this is necessary to enable attaching a `_begin` hook after construction
1413-
queueMicrotask(() => { iterator.read(); done(); });
1411+
scheduleTask(() => { iterator.read(); done(); });
14141412
});
14151413

14161414
it('should cause an exception', () => {
@@ -1637,7 +1635,7 @@ describe('BufferedIterator', () => {
16371635
before(() => {
16381636
iterator = new BufferedIterator();
16391637
iterator._begin = function (done) {
1640-
queueMicrotask(() => {
1638+
scheduleTask(() => {
16411639
this._push('x');
16421640
this._push('y');
16431641
done();
@@ -1947,7 +1945,7 @@ describe('BufferedIterator', () => {
19471945
done();
19481946
};
19491947
iterator._flush = function (done) {
1950-
queueMicrotask(() => {
1948+
scheduleTask(() => {
19511949
this._push('x');
19521950
this._push('y');
19531951
done();
@@ -2206,7 +2204,7 @@ describe('BufferedIterator', () => {
22062204
done();
22072205
};
22082206
iterator._flush = function (done) {
2209-
queueMicrotask(() => {
2207+
scheduleTask(() => {
22102208
this._push('x');
22112209
this._push('y');
22122210
done();

‎test/MultiTransformIterator-test.js

+7-7
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,10 @@ import {
66
EmptyIterator,
77
SingletonIterator,
88
ArrayIterator,
9+
scheduleTask,
910
} from '../asynciterator.mjs';
1011

1112
import { EventEmitter } from 'events';
12-
import queueMicrotask from 'queue-microtask';
1313

1414
describe('MultiTransformIterator', () => {
1515
describe('The MultiTransformIterator function', () => {
@@ -146,7 +146,7 @@ describe('MultiTransformIterator', () => {
146146
iterator = new MultiTransformIterator(source);
147147
iterator._createTransformer = sinon.spy(() => {
148148
const transformer = new BufferedIterator();
149-
setImmediate(() => transformer.close());
149+
setTimeout(() => transformer.close(), 0);
150150
return transformer;
151151
});
152152
});
@@ -181,7 +181,7 @@ describe('MultiTransformIterator', () => {
181181
iterator = new MultiTransformIterator(source);
182182
iterator._createTransformer = sinon.spy(item => {
183183
const transformer = new BufferedIterator();
184-
queueMicrotask(() => {
184+
scheduleTask(() => {
185185
transformer._push(`${item}1`);
186186
transformer.close();
187187
});
@@ -209,7 +209,7 @@ describe('MultiTransformIterator', () => {
209209
iterator = new MultiTransformIterator(source);
210210
iterator._createTransformer = sinon.spy(item => {
211211
const transformer = new BufferedIterator();
212-
queueMicrotask(() => {
212+
scheduleTask(() => {
213213
transformer._push(`${item}1`);
214214
transformer._push(`${item}2`);
215215
transformer._push(`${item}3`);
@@ -300,7 +300,7 @@ describe('MultiTransformIterator', () => {
300300
iterator = new MultiTransformIterator(source);
301301
iterator._createTransformer = sinon.spy(item => {
302302
const transformer = new BufferedIterator();
303-
queueMicrotask(() => {
303+
scheduleTask(() => {
304304
transformer.emit('error', new Error(`Error ${item}`));
305305
});
306306
return transformer;
@@ -319,7 +319,7 @@ describe('MultiTransformIterator', () => {
319319
source = new ArrayIterator(['a', 'b', 'c', 'd', 'e', 'f']);
320320
const multiTransform = sinon.spy(item => {
321321
const transformer = new BufferedIterator();
322-
queueMicrotask(() => {
322+
scheduleTask(() => {
323323
transformer._push(`${item}1`);
324324
transformer._push(`${item}2`);
325325
transformer._push(`${item}3`);
@@ -356,7 +356,7 @@ describe('MultiTransformIterator', () => {
356356
source = new ArrayIterator(['a', 'b', 'c', 'd', 'e', 'f']);
357357
const multiTransform = sinon.spy(item => {
358358
const transformer = new BufferedIterator();
359-
queueMicrotask(() => {
359+
scheduleTask(() => {
360360
transformer._push(`${item}1`);
361361
transformer._push(`${item}2`);
362362
transformer._push(`${item}3`);

‎test/SimpleTransformIterator-test.js

+2-2
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,10 @@ import {
66
EmptyIterator,
77
ArrayIterator,
88
IntegerIterator,
9+
scheduleTask,
910
} from '../asynciterator.mjs';
1011

1112
import { EventEmitter } from 'events';
12-
import queueMicrotask from 'queue-microtask';
1313

1414
describe('SimpleTransformIterator', () => {
1515
describe('The SimpleTransformIterator function', () => {
@@ -163,7 +163,7 @@ describe('SimpleTransformIterator', () => {
163163
source = new ArrayIterator(['a', 'b', 'c']);
164164
transform = sinon.spy(function (item, done) {
165165
this._push(item + (++i));
166-
queueMicrotask(done);
166+
scheduleTask(done);
167167
});
168168
iterator = new SimpleTransformIterator(source, transform);
169169
});

‎test/TaskScheduler-test.js

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
import {
2+
scheduleTask,
3+
getTaskScheduler,
4+
setTaskScheduler,
5+
} from '../asynciterator.mjs';
6+
7+
describe('TaskScheduler', () => {
8+
describe('scheduleTask', () => {
9+
it('is a function', () => {
10+
expect(scheduleTask).to.be.an.instanceof(Function);
11+
});
12+
13+
it('schedules a task', done => {
14+
scheduleTask(done);
15+
});
16+
});
17+
18+
describe('getTaskScheduler', () => {
19+
it('is a function', () => {
20+
expect(getTaskScheduler).to.be.an.instanceof(Function);
21+
});
22+
23+
it('returns a task scheduler', done => {
24+
const scheduler = getTaskScheduler();
25+
scheduler(done);
26+
});
27+
});
28+
29+
describe('setTaskScheduler', () => {
30+
it('is a function', () => {
31+
expect(setTaskScheduler).to.be.an.instanceof(Function);
32+
});
33+
34+
it('allows setting the task scheduler', () => {
35+
const scheduler = getTaskScheduler();
36+
expect(getTaskScheduler()).to.equal(scheduler);
37+
38+
const newScheduler = sinon.spy();
39+
setTaskScheduler(newScheduler);
40+
expect(getTaskScheduler()).to.equal(newScheduler);
41+
expect(newScheduler).to.not.have.been.called;
42+
43+
const task = sinon.spy();
44+
scheduleTask(task);
45+
expect(newScheduler).to.have.been.calledOnce;
46+
expect(newScheduler).to.have.been.calledWith(task);
47+
48+
setTaskScheduler(scheduler);
49+
});
50+
});
51+
});

‎test/TransformIterator-test.js

+3-3
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,10 @@ import {
55
ArrayIterator,
66
TransformIterator,
77
wrap,
8+
scheduleTask,
89
} from '../asynciterator.mjs';
910

1011
import { EventEmitter } from 'events';
11-
import queueMicrotask from 'queue-microtask';
1212

1313
describe('TransformIterator', () => {
1414
describe('The TransformIterator function', () => {
@@ -476,7 +476,7 @@ describe('TransformIterator', () => {
476476
before(() => {
477477
iterator = new TransformIterator(source = new ArrayIterator(['a', 'b', 'c']));
478478
iterator._transform = function (item, done) {
479-
queueMicrotask(() => {
479+
scheduleTask(() => {
480480
iterator._push(`${item}1`);
481481
iterator._push(`${item}2`);
482482
done();
@@ -588,7 +588,7 @@ describe('TransformIterator', () => {
588588
iterator = new TransformIterator(source);
589589
iterator._transform = sinon.spy(function (item, done) {
590590
this._push(item + (++i));
591-
queueMicrotask(done);
591+
scheduleTask(done);
592592
});
593593
});
594594

‎test/UnionIterator-test.js

+5-5
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,10 @@ import {
66
EmptyIterator,
77
union,
88
range,
9+
scheduleTask,
910
} from '../asynciterator.mjs';
1011

1112
import { EventEmitter } from 'events';
12-
import queueMicrotask from 'queue-microtask';
1313

1414
describe('UnionIterator', () => {
1515
describe('The UnionIterator function', () => {
@@ -96,7 +96,7 @@ describe('UnionIterator', () => {
9696
describe('after reading', () => {
9797
before(done => {
9898
iterator.read();
99-
queueMicrotask(done);
99+
scheduleTask(done);
100100
});
101101

102102
it('should have ended', () => {
@@ -168,7 +168,7 @@ describe('UnionIterator', () => {
168168
describe('after reading', () => {
169169
before(done => {
170170
iterator.read();
171-
queueMicrotask(done);
171+
scheduleTask(done);
172172
});
173173

174174
it('should have ended', () => {
@@ -405,7 +405,7 @@ describe('UnionIterator', () => {
405405
expect(iterator.read()).to.equal(7);
406406

407407
// Buffer
408-
await new Promise(resolve => queueMicrotask(resolve));
408+
await new Promise(resolve => scheduleTask(resolve));
409409

410410
// Read remaining items
411411
expect(iterator.read()).to.equal(5);
@@ -436,7 +436,7 @@ describe('UnionIterator', () => {
436436
const delayed = new AsyncIterator();
437437
const iterator = new UnionIterator(delayed);
438438
delayed.readable = true;
439-
queueMicrotask(() => delayed.close());
439+
scheduleTask(() => delayed.close());
440440
(await toArray(iterator)).should.eql([]);
441441
});
442442
});

0 commit comments

Comments
 (0)
Please sign in to comment.