use worker-rpc library for inter-process communication

package.json

@@ -87,7 +87,8 @@
     "micromatch": "^3.1.10",
     "minimatch": "^3.0.4",
     "semver": "^5.6.0",
-    "tapable": "^1.0.0"
+    "tapable": "^1.0.0",
+    "worker-rpc": "^0.1.0"
   },
   "husky": {
     "hooks": {

src/CancellationToken.ts

@@ -7,7 +7,7 @@ import * as ts from 'typescript'; // Imported for types alone
 
 import { FsHelper } from './FsHelper';
 
-interface CancellationTokenData {
+export interface CancellationTokenData {
   isCancelled: boolean;
   cancellationFileName: string;
 }

src/RpcTypes.ts

@@ -0,0 +1,9 @@
+import { CancellationTokenData } from './CancellationToken';
+import { Message } from './Message';
+
+export const RUN = 'run';
+export type RunPayload = CancellationTokenData;
+export type RunResult =
+  | Message
+  // when run was cancelled via CancellationToken, undefined is returned
+  | undefined;

src/cluster.ts

@@ -1,10 +1,16 @@
 import * as childProcess from 'child_process';
 import * as path from 'path';
 import * as process from 'process';
+import { RpcProvider } from 'worker-rpc';
 
-import { WorkResult } from './WorkResult';
 import { NormalizedMessage } from './NormalizedMessage';
 import { Message } from './Message';
+import { RunPayload, RunResult, RUN } from './RpcTypes';
+
+// helper function
+function noneUndefined<T>(results: (T | undefined)[]): results is T[] {
+  return results.every(result => !!result);
+}
 
 // fork workers...
 const division = parseInt(process.env.WORK_DIVISION || '', 10);
@@ -20,54 +26,58 @@ for (let num = 0; num < division; num++) {
   );
 }
 
-const pids = workers.map(worker => worker.pid);
-const result = new WorkResult(pids);
+// communication with parent process
+const parentRpc = new RpcProvider(message => {
+  try {
+    process.send!(message);
+  } catch (e) {
+    // channel closed...
+    process.exit();
+  }
+});
+process.on('message', message => parentRpc.dispatch(message));
 
-process.on('message', (message: Message) => {
-  // broadcast message to all workers
-  workers.forEach(worker => {
+// communication with worker processes
+const workerRpcs = workers.map(worker => {
+  const rpc = new RpcProvider(message => {
     try {
       worker.send(message);
     } catch (e) {
       // channel closed - something went wrong - close cluster...
       process.exit();
     }
   });
-
-  // clear previous result set
-  result.clear();
+  worker.on('message', message => rpc.dispatch(message));
+  return rpc;
 });
 
-// listen to all workers
-workers.forEach(worker => {
-  worker.on('message', (message: Message) => {
-    // set result from worker
-    result.set(worker.pid, {
-      diagnostics: message.diagnostics.map(NormalizedMessage.createFromJSON),
-      lints: message.lints.map(NormalizedMessage.createFromJSON)
-    });
+parentRpc.registerRpcHandler<RunPayload, RunResult>(RUN, async message => {
+  const workerResults = await Promise.all(
+    workerRpcs.map(workerRpc =>
+      workerRpc.rpc<RunPayload, RunResult>(RUN, message)
+    )
+  );
 
-    // if we have result from all workers, send merged
-    if (result.hasAll()) {
-      const merged: Message = result.reduce(
-        (innerMerged: Message, innerResult: Message) => ({
-          diagnostics: innerMerged.diagnostics.concat(innerResult.diagnostics),
-          lints: innerMerged.lints.concat(innerResult.lints)
-        }),
-        { diagnostics: [], lints: [] }
-      );
+  if (!noneUndefined(workerResults)) {
+    return undefined;
+  }
 
-      merged.diagnostics = NormalizedMessage.deduplicate(merged.diagnostics);
-      merged.lints = NormalizedMessage.deduplicate(merged.lints);
+  const merged: Message = workerResults.reduce(
+    (innerMerged: Message, innerResult: Message) => ({
+      diagnostics: innerMerged.diagnostics.concat(
+        innerResult.diagnostics.map(NormalizedMessage.createFromJSON)
+      ),
+      lints: innerMerged.lints.concat(
+        innerResult.lints.map(NormalizedMessage.createFromJSON)
+      )
+    }),
+    { diagnostics: [], lints: [] }
+  );
 
-      try {
-        process.send!(merged);
-      } catch (e) {
-        // channel closed...
-        process.exit();
-      }
-    }
-  });
+  merged.diagnostics = NormalizedMessage.deduplicate(merged.diagnostics);
+  merged.lints = NormalizedMessage.deduplicate(merged.lints);
+
+  return merged;
 });
 
 process.on('SIGINT', () => {

src/index.ts

@@ -1,6 +1,7 @@
 import * as path from 'path';
 import * as process from 'process';
 import * as childProcess from 'child_process';
+import { RpcProvider } from 'worker-rpc';
 import * as semver from 'semver';
 import chalk, { Chalk } from 'chalk';
 import * as micromatch from 'micromatch';
@@ -17,6 +18,7 @@ import { FsHelper } from './FsHelper';
 import { Message } from './Message';
 
 import { getForkTsCheckerWebpackPluginHooks, legacyHookMap } from './hooks';
+import { RunPayload, RunResult, RUN } from './RpcTypes';
 
 const checkerPluginName = 'fork-ts-checker-webpack-plugin';
 
@@ -121,6 +123,7 @@ class ForkTsCheckerWebpackPlugin {
   private tslintVersion: string;
 
   private service?: childProcess.ChildProcess;
+  private serviceRpc?: RpcProvider;
 
   private vue: boolean;
 
@@ -395,7 +398,14 @@ class ForkTsCheckerWebpackPlugin {
             if (this.measureTime) {
               this.startAt = this.performance.now();
             }
-            this.service!.send(this.cancellationToken);
+            this.serviceRpc!.rpc<RunPayload, RunResult>(
+              RUN,
+              this.cancellationToken.toJSON()
+            ).then(result => {
+              if (result) {
+                this.handleServiceMessage(result);
+              }
+            });
           } catch (error) {
             if (!this.silent && this.logger) {
               this.logger.error(
@@ -578,6 +588,8 @@ class ForkTsCheckerWebpackPlugin {
         stdio: ['inherit', 'inherit', 'inherit', 'ipc']
       }
     );
+    this.serviceRpc = new RpcProvider(message => this.service!.send(message));
+    this.service.on('message', message => this.serviceRpc!.dispatch(message));
 
     if ('hooks' in this.compiler) {
       // webpack 4+
@@ -630,9 +642,6 @@ class ForkTsCheckerWebpackPlugin {
       }
     }
 
-    this.service.on('message', (message: Message) =>
-      this.handleServiceMessage(message)
-    );
     this.service.on('exit', (code: string | number, signal: string) =>
       this.handleServiceExit(code, signal)
     );
@@ -649,6 +658,7 @@ class ForkTsCheckerWebpackPlugin {
 
       this.service.kill();
       this.service = undefined;
+      this.serviceRpc = undefined;
     } catch (e) {
       if (this.logger && !this.silent) {
         this.logger.error(e);

src/service.ts

@@ -10,6 +10,18 @@ import {
   makeCreateNormalizedMessageFromDiagnostic,
   makeCreateNormalizedMessageFromRuleFailure
 } from './NormalizedMessageFactories';
+import { RpcProvider } from 'worker-rpc';
+import { RunPayload, RunResult, RUN } from './RpcTypes';
+
+const rpc = new RpcProvider(message => {
+  try {
+    process.send!(message);
+  } catch (e) {
+    // channel closed...
+    process.exit();
+  }
+});
+process.on('message', message => rpc.dispatch(message));
 
 const typescript: typeof ts = require(process.env.TYPESCRIPT_PATH!);
 
@@ -61,28 +73,25 @@ async function run(cancellationToken: CancellationToken) {
     }
   } catch (error) {
     if (error instanceof typescript.OperationCanceledException) {
-      return;
+      return undefined;
     }
 
     throw error;
   }
 
-  if (!cancellationToken.isCancellationRequested()) {
-    try {
-      process.send!({
-        diagnostics,
-        lints
-      });
-    } catch (e) {
-      // channel closed...
-      process.exit();
-    }
+  if (cancellationToken.isCancellationRequested()) {
+    return undefined;
   }
+
+  return {
+    diagnostics,
+    lints
+  };
 }
 
-process.on('message', message => {
-  run(CancellationToken.createFromJSON(typescript, message));
-});
+rpc.registerRpcHandler<RunPayload, RunResult>(RUN, message =>
+  run(CancellationToken.createFromJSON(typescript, message!))
+);
 
 process.on('SIGINT', () => {
   process.exit();

yarn.lock

@@ -2207,6 +2207,10 @@ merge-source-map@^1.1.0:
   dependencies:
     source-map "^0.6.1"
 
+microevent.ts@~0.1.0:
+  version "0.1.0"
+  resolved "https://registry.yarnpkg.com/microevent.ts/-/microevent.ts-0.1.0.tgz#390748b8a515083e6b63cd5112a3f18c2fe0eba8"
+
 micromatch@^3.1.10, micromatch@^3.1.4, micromatch@^3.1.8:
   version "3.1.10"
   resolved "https://registry.yarnpkg.com/micromatch/-/micromatch-3.1.10.tgz#70859bc95c9840952f359a068a3fc49f9ecfac23"
@@ -3693,7 +3697,6 @@ typedarray@^0.0.6:
 typescript@^3.0.1:
   version "3.3.3333"
   resolved "https://registry.yarnpkg.com/typescript/-/typescript-3.3.3333.tgz#171b2c5af66c59e9431199117a3bcadc66fdcfd6"
-  integrity sha512-JjSKsAfuHBE/fB2oZ8NxtRTk5iGcg6hkYXMnZ3Wc+b2RSqejEqTaem11mHASMnFilHrax3sLK0GDzcJrekZYLw==
 
 uglify-js@^3.1.4:
   version "3.4.9"
@@ -3734,7 +3737,6 @@ unique-slug@^2.0.0:
 unixify@^1.0.0:
   version "1.0.0"
   resolved "https://registry.yarnpkg.com/unixify/-/unixify-1.0.0.tgz#3a641c8c2ffbce4da683a5c70f03a462940c2090"
-  integrity sha1-OmQcjC/7zk2mg6XHDwOkYpQMIJA=
   dependencies:
     normalize-path "^2.1.1"
 
@@ -3892,6 +3894,12 @@ worker-farm@^1.5.2:
   dependencies:
     errno "~0.1.7"
 
+worker-rpc@^0.1.0:
+  version "0.1.0"
+  resolved "https://registry.yarnpkg.com/worker-rpc/-/worker-rpc-0.1.0.tgz#5f1258dca3d617cd18ca86587f8a05ac0eebd834"
+  dependencies:
+    microevent.ts "~0.1.0"
+
 wrap-ansi@^3.0.1:
   version "3.0.1"
   resolved "https://registry.yarnpkg.com/wrap-ansi/-/wrap-ansi-3.0.1.tgz#288a04d87eda5c286e060dfe8f135ce8d007f8ba"