stopMessagePort

} = internalBinding('messaging');

const {

threadId,

getEnvMessagePort

} = internalBinding('worker');

const { Readable, Writable } = require('stream');

const EventEmitter = require('events');

const {

Event,

NodeEventTarget,

defineEventHandler,

initNodeEventTarget,

kCreateEvent,

kNewListener,

kRemoveListener,

} = require('internal/event_target');

const { inspect } = require('internal/util/inspect');

let debug = require('internal/util/debuglog').debuglog('worker', (fn) => {

debug = fn;

});

const kIncrementsPortRef = Symbol('kIncrementsPortRef');

const kName = Symbol('kName');

const kOnMessageListener = Symbol('kOnMessageListener');

const kPort = Symbol('kPort');

const kWaitingStreams = Symbol('kWaitingStreams');

const kWritableCallbacks = Symbol('kWritableCallbacks');

};

// We have to mess with the MessagePort prototype a bit, so that a) we can make

// it inherit from EventEmitter, even though it is a C++ class, and b) we do

// it inherit from NodeEventTarget, even though it is a C++ class, and b) we do

// not provide methods that are not present in the Browser and not documented

// on our side (e.g. hasRef).

// Save a copy of the original set of methods as a shallow clone.

const MessagePortPrototype = ObjectCreate(

ObjectGetPrototypeOf(MessagePort.prototype),

ObjectGetOwnPropertyDescriptors(MessagePort.prototype));

// Set up the new inheritance chain.

ObjectSetPrototypeOf(MessagePort, EventEmitter);

ObjectSetPrototypeOf(MessagePort.prototype, EventEmitter.prototype);

ObjectSetPrototypeOf(MessagePort, NodeEventTarget);

ObjectSetPrototypeOf(MessagePort.prototype, NodeEventTarget.prototype);

// Copy methods that are inherited from HandleWrap, because

// changing the prototype of MessagePort.prototype implicitly removed them.

MessagePort.prototype.ref = MessagePortPrototype.ref;

MessagePort.prototype.unref = MessagePortPrototype.unref;

// A communication channel consisting of a handle (that wraps around an

// uv_async_t) which can receive information from other threads and emits

// .onmessage events, and a function used for sending data to a MessagePort

// in some other thread.

MessagePort.prototype[kOnMessageListener] = function onmessage(event) {

if (event.data && event.data.type !== messageTypes.STDIO_WANTS_MORE_DATA)

debug(`[${threadId}] received message`, event);

// Emit the deserialized object to userland.

this.emit('message', event.data);

};

// This is for compatibility with the Web's MessagePort API. It makes sense to

// provide it as an `EventEmitter` in Node.js, but if somebody overrides

// `onmessage`, we'll switch over to the Web API model.

ObjectDefineProperty(MessagePort.prototype, 'onmessage', {

enumerable: true,

configurable: true,

get() {

return this[kOnMessageListener];

},

set(value) {

this[kOnMessageListener] = value;

if (typeof value === 'function') {

this.ref();

MessagePortPrototype.start.call(this);

} else {

this.unref();

stopMessagePort(this);

}

class MessageEvent extends Event {

constructor(data, target, type) {

super(type);

this.data = data;

}

});

}

ObjectDefineProperty(

MessagePort.prototype,

kCreateEvent,

{

value: function(data, type) {

return new MessageEvent(data, this, type);

},

configurable: false,

writable: false,

enumerable: false,

});

// This is called from inside the `MessagePort` constructor.

function oninit() {

initNodeEventTarget(this);

// TODO(addaleax): This should be on MessagePort.prototype, but

// defineEventHandler() does not support that.

defineEventHandler(this, 'message');

defineEventHandler(this, 'messageerror');

setupPortReferencing(this, this, 'message');

}

value: oninit

});

class MessagePortCloseEvent extends Event {

constructor() {

super('close');

}

}

// This is called after the underlying `uv_async_t` has been closed.

function onclose() {

this.emit('close');

this.dispatchEvent(new MessagePortCloseEvent());

}

ObjectDefineProperty(MessagePort.prototype, handleOnCloseSymbol, {

// If there are none or all are removed, unref() the channel so the worker

// can shutdown gracefully.

port.unref();

eventEmitter.on('newListener', (name) => {

if (name === eventName && eventEmitter.listenerCount(eventName) === 0) {

eventEmitter.on('newListener', function(name) {

if (name === eventName) newListener(eventEmitter.listenerCount(name));

});

eventEmitter.on('removeListener', function(name) {

if (name === eventName) removeListener(eventEmitter.listenerCount(name));

});

const origNewListener = eventEmitter[kNewListener];

eventEmitter[kNewListener] = function(size, type, ...args) {

if (type === eventName) newListener(size - 1);

return origNewListener.call(this, size, type, ...args);

};

const origRemoveListener = eventEmitter[kRemoveListener];

eventEmitter[kRemoveListener] = function(size, type, ...args) {

if (type === eventName) removeListener(size);

return origRemoveListener.call(this, size, type, ...args);

};

function newListener(size) {

if (size === 0) {

port.ref();

MessagePortPrototype.start.call(port);

}

});

eventEmitter.on('removeListener', (name) => {

if (name === eventName && eventEmitter.listenerCount(eventName) === 0) {

}

function removeListener(size) {

if (size === 0) {

stopMessagePort(port);

port.unref();

}

});

}

}