maxmustermann/kibana
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Revert "[kbn/optimizer] remove disconnect listener (#67161)"

Browse source 
This reverts commit 90fc521d21.
This commit is contained in:
spalger 2020-05-21 20:52:43 -07:00
parent fd7a9a7e13
commit 556ae523c8
2 changed files with 36 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
packages/kbn-optimizer/src/worker/observe_parent_offline.test.ts
View file

@ -75,6 +75,18 @@ async function waitForTick() {
}
describe('emits and completes when parent exists because:', () => {
test('"disconnect" event', async () => {
const mockProc = new MockProcess();
const promise = record(observeParentOffline(mockProc, workerMsgs));
mockProc.emit('disconnect');
expect(await promise).toMatchInlineSnapshot(`
Array [
"next: 'parent offline (disconnect event)'",
"complete",
]
`);
});
test('process.connected is false', async () => {
const mockProc = new MockProcess({
connected: false,

41
packages/kbn-optimizer/src/worker/observe_parent_offline.ts
View file

@ -44,28 +44,35 @@ export interface Process extends EventEmitter {
* call errored with an 'ERR_IPC_CHANNEL_CLOSED' exception
*/
export function observeParentOffline(process: Process, workerMsgs: WorkerMsgs) {
return sleep(5000).pipe(
mergeMap(() => {
if (!process.connected || !process.send) {
return Rx.of('parent offline (disconnected)');
}
return Rx.race(
Rx.fromEvent(process, 'disconnect').pipe(
take(1),
map(() => 'parent offline (disconnect event)')
),
process.send(workerMsgs.ping());
sleep(5000).pipe(
mergeMap(() => {
if (!process.connected || !process.send) {
return Rx.of('parent offline (disconnected)');
}
const pong$ = Rx.fromEvent<[any]>(process, 'message').pipe(
first(([msg]) => isParentPong(msg)),
map(() => {
throw new Error('parent still online');
})
);
process.send(workerMsgs.ping());
// give the parent some time to respond, if the ping
// wins the race the parent is considered online
const timeout$ = sleep(5000).pipe(map(() => 'parent offline (ping timeout)'));
const pong$ = Rx.fromEvent<[any]>(process, 'message').pipe(
first(([msg]) => isParentPong(msg)),
map(() => {
throw new Error('parent still online');
})
);
return Rx.race(pong$, timeout$);
}),
// give the parent some time to respond, if the ping
// wins the race the parent is considered online
const timeout$ = sleep(5000).pipe(map(() => 'parent offline (ping timeout)'));
return Rx.race(pong$, timeout$);
})
)
).pipe(
/**
* resubscribe to the source observable (triggering the timer,
* ping, wait for response) if the source observable does not