Spaces:
Sleeping
Sleeping
File size: 1,918 Bytes
b593f0b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | // @flow
import {uniqueId, asyncAll} from './util';
import Actor from './actor';
import assert from 'assert';
import type WorkerPool from './worker_pool';
/**
* Responsible for sending messages from a {@link Source} to an associated
* {@link WorkerSource}.
*
* @private
*/
class Dispatcher {
workerPool: WorkerPool;
actors: Array<Actor>;
currentActor: number;
id: number;
// exposed to allow stubbing in unit tests
static Actor: Class<Actor>;
constructor(workerPool: WorkerPool, parent: any) {
this.workerPool = workerPool;
this.actors = [];
this.currentActor = 0;
this.id = uniqueId();
const workers = this.workerPool.acquire(this.id);
for (let i = 0; i < workers.length; i++) {
const worker = workers[i];
const actor = new Dispatcher.Actor(worker, parent, this.id);
actor.name = `Worker ${i}`;
this.actors.push(actor);
}
assert(this.actors.length);
}
/**
* Broadcast a message to all Workers.
* @private
*/
broadcast(type: string, data: mixed, cb?: Function) {
assert(this.actors.length);
cb = cb || function () {};
asyncAll(this.actors, (actor, done) => {
actor.send(type, data, done);
}, cb);
}
/**
* Acquires an actor to dispatch messages to. The actors are distributed in round-robin fashion.
* @returns An actor object backed by a web worker for processing messages.
*/
getActor(): Actor {
assert(this.actors.length);
this.currentActor = (this.currentActor + 1) % this.actors.length;
return this.actors[this.currentActor];
}
remove() {
this.actors.forEach((actor) => { actor.remove(); });
this.actors = [];
this.workerPool.release(this.id);
}
}
Dispatcher.Actor = Actor;
export default Dispatcher;
|