Handle empty lines in Docker event streams

Author: pimterry

custom-typings/event-stream.d.ts

@@ -1,5 +1,5 @@
 declare module 'event-stream' {
-    // Type definitions for event-stream v3.3.2
+    // Type definitions for event-stream v4.0
     // Project: https://github.com/dominictarr/event-stream
     // Definitions by: David Gardiner <https://github.com/flcdrg>
     // Definitions: https://github.com/DefinitelyTyped/DefinitelyTyped
@@ -8,7 +8,7 @@ declare module 'event-stream' {
 
     import * as stream from 'stream';
 
-    interface MapStream extends stream.Stream {
+    export interface MapStream extends stream.Stream {
         writable: boolean;
         readable: boolean;
 
@@ -22,7 +22,7 @@ declare module 'event-stream' {
     /**
      * Create a stream from a series of MapStreams
      */
-    export function pipeline(...streams: MapStream[]): NodeJS.ReadWriteStream;
+    export function pipeline(...streams: MapStream[]): MapStream;
 
     /**
      * Create a through stream from an asynchronous function
@@ -37,6 +37,11 @@ declare module 'event-stream' {
      */
     export function mapSync(syncFunction: Function):  MapStream;
 
+    /**
+     * Filter elements of the stream.
+     */
+    export function filterSync(syncFunction: (value: any) => boolean):  MapStream;
+
     /**
      * Break up a stream and reassemble it so that each line is a chunk. matcher may be a String, or a RegExp
      *

src/interceptors/docker/docker-build-injection.ts

@@ -218,7 +218,7 @@ const END_OF_STEP_REGEX = /^ ---\u003e [a-z0-9]+\n$/;
  * HTTP Toolkit interception noise down to a single clear line, and normalize the step count text
  * to match.
  */
-export function getBuildOutputPipeline(extraDockerCommandCount: number): NodeJS.ReadWriteStream {
+export function getBuildOutputPipeline(extraDockerCommandCount: number): EventStream.MapStream {
     let outputToHide: 'none' | 'all' | 'until-next' = 'none';
 
     let totalStepCount = -1;

src/interceptors/docker/docker-networking.ts

@@ -20,7 +20,7 @@ interface DockerEvent {
     };
 }
 
-let dockerEventStream: NodeJS.ReadableStream | undefined;
+let dockerEventStream: EventStream.MapStream | undefined;
 
 /**
  * Get a stream of events from Docker. Although Dockerode will only give us the raw event buffers,
@@ -33,31 +33,29 @@ function getDockerEventStream(docker: Docker) {
     if (!dockerEventStream) {
         const dockerEventParsingStream = EventStream.pipeline(
             EventStream.split(),
-            EventStream.mapSync((rawLine: Buffer) =>
-                JSON.parse(rawLine.toString('utf8'))
-            )
+            EventStream.mapSync((buffer: Buffer) => buffer.toString('utf8')),
+            EventStream.filterSync((line: string) => line.length > 0),
+            EventStream.mapSync((rawLine: string) => JSON.parse(rawLine))
         );
 
         // We expose the stream immediately, even though no data is coming yet
         dockerEventStream = dockerEventParsingStream;
 
         // This gives us a stream of raw Buffer data. Inside, it contains
         // JSON strings, newline separated, which we parse above.
-        (docker.getEvents() as Promise<stream.Readable>) // Wrong types, it's a raw http.IncomingMessage
-            .then((rawEventStream) => {
-                rawEventStream.pipe(dockerEventParsingStream);
-                rawEventStream.on('error', (e) => {
-                    dockerEventParsingStream?.emit('error', e);
-                });
-                rawEventStream.on('close', () => {
-                    dockerEventParsingStream?.end();
-                    dockerEventStream = undefined;
-                });
-            })
-            .catch((e) => {
+        docker.getEvents().then((rawEventStream) => {
+            rawEventStream.pipe(dockerEventParsingStream);
+            rawEventStream.on('error', (e) => {
                 dockerEventParsingStream?.emit('error', e);
+            });
+            rawEventStream.on('close', () => {
+                dockerEventParsingStream?.end();
                 dockerEventStream = undefined;
             });
+        }).catch((e) => {
+            dockerEventParsingStream?.emit('error', e);
+            dockerEventStream = undefined;
+        });
     }
 
     return dockerEventStream;
@@ -166,7 +164,7 @@ class DockerNetworkMonitor {
     constructor(
         private docker: Docker,
         private proxyPort: number,
-        private dockerEventStream: NodeJS.ReadableStream
+        private dockerEventStream: stream.Stream
     ) {
         // We use mobx here to automatically propagate updates whilst avoiding
         // unnecessary updates when nothing changes.