Skip to content

Commit

Permalink
child_process: queue pending messages
Browse files Browse the repository at this point in the history
It fixes the problem of the child process not receiving messages.

Fixes: #41134

PR-URL: #41221
Reviewed-By: Adrian Estrada <edsadr@gmail.com>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
Reviewed-By: Minwoo Jung <nodecorelab@gmail.com>
  • Loading branch information
ErickWendel authored and targos committed Jan 14, 2022
1 parent 75c565b commit d9465ae
Show file tree
Hide file tree
Showing 2 changed files with 49 additions and 1 deletion.
30 changes: 29 additions & 1 deletion lib/internal/child_process.js
Expand Up @@ -81,6 +81,7 @@ let HTTPParser;
const MAX_HANDLE_RETRANSMISSIONS = 3;
const kChannelHandle = Symbol('kChannelHandle');
const kIsUsedAsStdio = Symbol('kIsUsedAsStdio');
const kPendingMessages = Symbol('kPendingMessages');

// This object contain function to convert TCP objects to native handle objects
// and back again.
Expand Down Expand Up @@ -526,6 +527,7 @@ class Control extends EventEmitter {
constructor(channel) {
super();
this.#channel = channel;
this[kPendingMessages] = [];
}

// The methods keeping track of the counter are being used to track the
Expand Down Expand Up @@ -699,6 +701,24 @@ function setupChannel(target, channel, serializationMode) {
});
});

target.on('newListener', function() {

process.nextTick(() => {
if (!target.channel || !target.listenerCount('message'))
return;

const messages = target.channel[kPendingMessages];
const { length } = messages;
if (!length) return;

for (let i = 0; i < length; i++) {
ReflectApply(target.emit, target, messages[i]);
}

target.channel[kPendingMessages] = [];
});
});

target.send = function(message, handle, options, callback) {
if (typeof handle === 'function') {
callback = handle;
Expand Down Expand Up @@ -912,7 +932,15 @@ function setupChannel(target, channel, serializationMode) {
};

function emit(event, message, handle) {
target.emit(event, message, handle);
if ('internalMessage' === event || target.listenerCount('message')) {
target.emit(event, message, handle);
return;
}

ArrayPrototypePush(
target.channel[kPendingMessages],
[event, message, handle]
);
}

function handleMessage(message, handle, internal) {
Expand Down
20 changes: 20 additions & 0 deletions test/es-module/test-esm-child-process-fork-main.mjs
@@ -0,0 +1,20 @@
import '../common/index.mjs';
import assert from 'assert';
import { fork } from 'child_process';
import { once } from 'events';
import { fileURLToPath } from 'url';

if (process.argv[2] !== 'child') {
const filename = fileURLToPath(import.meta.url);
const cp = fork(filename, ['child']);
const message = 'Hello World';
cp.send(message);

const [received] = await once(cp, 'message');
assert.deepStrictEqual(received, message);

cp.disconnect();
await once(cp, 'exit');
} else {
process.on('message', (msg) => process.send(msg));
}

0 comments on commit d9465ae

Please sign in to comment.