fix(core): When piping a stream to multiple destinations, do so safely

zenflow · zenflow · commit 0d32c17871c5 · 2020-07-16T18:26:50.000-04:00
diff --git a/package.json b/package.json
@@ -37,7 +37,7 @@
     "_test": "tsdx test --runInBand",
     "test": "yarn _test --ci",
     "test-watch": "yarn _test --watch",
-    "lint": "tsdx lint scripts src test types website",
+    "lint": "tsdx lint scripts src test website",
     "lint-fix": "yarn lint --fix",
     "extract-api": "api-extractor run --local",
     "check-api": "api-extractor run",
@@ -49,6 +49,7 @@
   },
   "dependencies": {
     "@scarf/scarf": "^1.0.6",
+    "cloneable-readable": "^2.0.1",
     "express": "^4.17.1",
     "http-proxy-middleware": "^1.0.4",
     "merge-stream": "^2.0.0",
@@ -64,6 +65,7 @@
     "@microsoft/api-extractor": "^7.9.0",
     "@semantic-release/changelog": "^5.0.1",
     "@semantic-release/git": "^9.0.0",
+    "@types/cloneable-readable": "^2.0.0",
     "@types/express": "^4.17.6",
     "@types/merge-stream": "^1.1.2",
     "@types/node-fetch": "^2.5.7",
diff --git a/src/core/Service.ts b/src/core/Service.ts
@@ -1,4 +1,5 @@
 import { PassThrough } from 'stream'
+import cloneable from 'cloneable-readable'
 import { ServiceProcess } from './ServiceProcess'
 import { NormalizedServiceConfig } from './validateAndNormalizeConfig'
 import { ReadyContext } from './ReadyContext'
@@ -10,9 +11,10 @@ import { Logger } from './Logger'
 export class Service {
   public readonly id: string
   public readonly config: NormalizedServiceConfig
-  public readonly output = new PassThrough({ objectMode: true })
+  public readonly output = cloneable(new PassThrough({ objectMode: true }))
   private readonly logger: Logger
   private readonly die: (message: string) => Promise<never>
+  private readonly outputClone = this.output.clone()
   private ready: Promise<void> | undefined
   private process: ServiceProcess | undefined
   private startResult: Promise<void> | undefined
@@ -46,11 +48,16 @@ export class Service {
     return this.startResult
   }
   private defineReady() {
-    const output = this.output.pipe(new PassThrough({ objectMode: true }))
-    const ctx: ReadyContext = { output }
-    this.ready = promiseTry(() => this.config.ready(ctx)).catch(error =>
-      this.die(`Error from ready function: ${maybeErrorText(error)}`)
-    )
+    const ctx: ReadyContext = {
+      output: this.outputClone,
+    }
+    this.ready = promiseTry(() => this.config.ready(ctx))
+      .catch(error =>
+        this.die(`Error from ready function: ${maybeErrorText(error)}`)
+      )
+      .then(() => {
+        this.outputClone.destroy()
+      })
   }
   private async startProcess() {
     const proc = new ServiceProcess(this.config, () => {
@@ -103,7 +110,7 @@ export class Service {
   }
   public stop() {
     if (!this.stopResult) {
-      if (!this.process || this.process.isEnded) {
+      if (!this.process || !this.process.isRunning()) {
         this.stopResult = Promise.resolve()
       } else {
         this.logger.info(`Stopping service '${this.id}'...`)
diff --git a/src/core/ServiceProcess.ts b/src/core/ServiceProcess.ts
@@ -1,70 +1,75 @@
-import { PassThrough } from 'stream'
+import { Readable, PassThrough } from 'stream'
 import { ChildProcessWithoutNullStreams } from 'child_process'
-import { once } from 'events'
 import mergeStream from 'merge-stream'
 import splitStream from 'split'
 import { NormalizedServiceConfig } from './validateAndNormalizeConfig'
 import { spawnProcess } from './spawnProcess'
 
-const split = () => splitStream((line: string) => `${line}\n`)
-
 export class ServiceProcess {
   public readonly output = new PassThrough({ objectMode: true })
   public readonly started: Promise<void>
-  public isEnded = false
   public logTail: string[] = []
   private readonly process: ChildProcessWithoutNullStreams
+  private didError = false
+  private didEnd = false
   private readonly ended: Promise<void>
   private wasEndCalled = false
   constructor(config: NormalizedServiceConfig, onCrash: () => void) {
     this.process = spawnProcess(config)
-    const childOutput = mergeStream(
-      this.process.stdout.setEncoding('utf8').pipe(split()),
-      this.process.stderr.setEncoding('utf8').pipe(split())
-    )
-    childOutput.pipe(this.output)
-    if (config.logTailLength > 0) {
-      this.output.on('data', line => {
-        this.logTail.push(line)
-        if (this.logTail.length > config.logTailLength) {
-          this.logTail.shift()
-        }
-      })
-    }
-    const error = new Promise(resolve => this.process.on('error', resolve))
     this.started = Promise.race([
-      error,
-      new Promise(resolve => setTimeout(resolve, 100)),
+      new Promise(resolve => this.process.once('error', resolve)),
+      new Promise(resolve => setTimeout(() => resolve(), 100)),
     ]).then(error => {
-      if (!error) {
-        return
+      if (error) {
+        this.didError = true
+        throw error
       }
-      childOutput.unpipe(this.output)
-      this.output.end()
-      return Promise.reject(error)
     })
-    const didStart = this.started.then(
-      () => true,
-      () => false
+    const processOutput = mergeStream(
+      transformStream(this.process.stdout),
+      transformStream(this.process.stderr)
     )
-    this.ended = Promise.race([error, once(childOutput, 'end')]).then(() => {
-      this.isEnded = true
-      if (!this.wasEndCalled) {
-        didStart.then(didStart => {
-          if (didStart) {
-            onCrash()
+    this.ended = (async () => {
+      for await (const line of processOutput as AsyncIterable<string>) {
+        if (this.didError) {
+          break
+        }
+        this.output.write(line)
+        if (config.logTailLength > 0) {
+          this.logTail.push(line)
+          if (this.logTail.length > config.logTailLength) {
+            this.logTail.shift()
           }
-        })
+        }
+      }
+      this.didEnd = true
+      this.output.end()
+    })()
+    Promise.all([this.started.catch(() => {}), this.ended]).then(() => {
+      if (!this.didError && !this.wasEndCalled) {
+        onCrash()
       }
     })
   }
-  end() {
+  public isRunning() {
+    return !this.didError && !this.didEnd
+  }
+  public end() {
     if (!this.wasEndCalled) {
       this.wasEndCalled = true
-      if (!this.isEnded) {
+      if (this.isRunning()) {
         this.process.kill('SIGINT')
       }
     }
     return this.ended
   }
 }
+
+/**
+ * Split input into stream of utf8 strings ending in '\n'
+ * */
+function transformStream(input: Readable): Readable {
+  return input
+    .setEncoding('utf8')
+    .pipe(splitStream((line: string) => `${line}\n`))
+}
diff --git a/yarn.lock b/yarn.lock
@@ -1372,6 +1372,13 @@
     "@types/connect" "*"
     "@types/node" "*"
 
+"@types/cloneable-readable@^2.0.0":
+  version "2.0.0"
+  resolved "https://registry.yarnpkg.com/@types/cloneable-readable/-/cloneable-readable-2.0.0.tgz#b5bc6602d4771a5db9d80f3867427e9da5f68a63"
+  integrity sha512-Q9fTsA3hEbOXmGIZ7StMunr/SCvtdfXDfJcDadYk/MPbS3Xh/fWCsdhW26NVx1XNNcX3SkdBqPkfbOiD6p3q2Q==
+  dependencies:
+    "@types/node" "*"
+
 "@types/color-name@^1.1.1":
   version "1.1.1"
   resolved "https://registry.yarnpkg.com/@types/color-name/-/color-name-1.1.1.tgz#1c1261bbeaa10a8055bbc5d8ab84b7b2afc846a0"
@@ -2613,6 +2620,14 @@ clone@^1.0.2:
   resolved "https://registry.yarnpkg.com/clone/-/clone-1.0.4.tgz#da309cc263df15994c688ca902179ca3c7cd7c7e"
   integrity sha1-2jCcwmPfFZlMaIypAheco8fNfH4=
 
+cloneable-readable@^2.0.1:
+  version "2.0.1"
+  resolved "https://registry.yarnpkg.com/cloneable-readable/-/cloneable-readable-2.0.1.tgz#fc2240beddbe5621b872acad8104dcc86574e225"
+  integrity sha512-1ke/wckhpSevGPQzKb+qGHMsuFrSUKQlsKh0PTmscmfAzw8MgONqrg5a0e0Un1YO/cOSS4wAepfXSGus5RoonQ==
+  dependencies:
+    inherits "^2.0.1"
+    readable-stream "^3.3.0"
+
 cmd-shim@^3.0.0, cmd-shim@^3.0.3:
   version "3.0.3"
   resolved "https://registry.yarnpkg.com/cmd-shim/-/cmd-shim-3.0.3.tgz#2c35238d3df37d98ecdd7d5f6b8dc6b21cadc7cb"
@@ -7708,7 +7723,7 @@ read@1, read@~1.0.1, read@~1.0.7:
     string_decoder "~1.1.1"
     util-deprecate "~1.0.1"
 
-"readable-stream@2 || 3", readable-stream@^3.6.0:
+"readable-stream@2 || 3", readable-stream@^3.3.0, readable-stream@^3.6.0:
   version "3.6.0"
   resolved "https://registry.yarnpkg.com/readable-stream/-/readable-stream-3.6.0.tgz#337bbda3adc0706bd3e024426a286d4b4b2c9198"
   integrity sha512-BViHy7LKeTz4oNnkcLJ+lVSL6vpiFeX6/d3oSH8zCW7UxP2onchk+vTGB143xuFjHS3deTgkKoXXymXqymiIdA==
