Fix livepush to work with node 16

This also improves the memory efficiency of the sync mechanism by
calculating the stage ids on the fly instead of storing the full
build output in memory and then parsing the string.

sync/device.ts

@@ -5,8 +5,9 @@ import { Builder } from 'resin-docker-build';
 
 import { promises as fs } from 'fs';
 import * as Path from 'path';
-import { Duplex, Readable, PassThrough, Stream } from 'stream';
+import { Readable } from 'stream';
 import * as tar from 'tar-stream';
+import * as readline from 'readline';
 
 import { exec } from '../src/lib/fs-utils';
 
@@ -52,7 +53,7 @@ export async function getDeviceArch(docker: Docker): Promise<string> {
 		}
 
 		return arch.trim();
-	} catch (e) {
+	} catch (e: any) {
 		throw new Error(
 			`Unable to get device architecture: ${e.message}.\nTry specifying the architecture with -a.`,
 		);
@@ -68,31 +69,61 @@ export async function getCacheFrom(docker: Docker): Promise<string[]> {
 	}
 }
 
+// Source: https://github.com/balena-io/balena-cli/blob/f6d668684a6f5ea8102a964ca1942b242eaa7ae2/lib/utils/device/live.ts#L539-L547
+function extractDockerArrowMessage(outputLine: string): string | undefined {
+	const arrowTest = /^.*\s*-+>\s*(.+)/i;
+	const match = arrowTest.exec(outputLine);
+	if (match != null) {
+		return match[1];
+	}
+}
+
 export async function performBuild(
 	docker: Docker,
 	dockerfile: Dockerfile,
 	dockerOpts: { [key: string]: any },
-): Promise<string> {
+): Promise<string[]> {
 	const builder = Builder.fromDockerode(docker);
 
 	// tar the directory, but replace the dockerfile with the
 	// livepush generated one
 	const tarStream = await tarDirectory(Path.join(__dirname, '..'), dockerfile);
-	const bufStream = new PassThrough();
 
 	return new Promise((resolve, reject) => {
-		const chunks = [] as Buffer[];
+		// Store the stage ids for caching
-		bufStream.on('data', (chunk) => chunks.push(Buffer.from(chunk)));
+		const ids = [] as string[];
 		builder.createBuildStream(dockerOpts, {
 			buildSuccess: () => {
-				// Return the build logs
+				// Return the image ids
-				resolve(Buffer.concat(chunks).toString('utf8'));
+				resolve(ids);
 			},
 			buildFailure: reject,
-			buildStream: (stream: Duplex) => {
+			buildStream: (input: NodeJS.ReadWriteStream) => {
-				stream.pipe(process.stdout);
+				// Parse the build output to get stage ids and
-				stream.pipe(bufStream);
+				// for logging
-				tarStream.pipe(stream);
+				let lastArrowMessage: string | undefined;
+				readline.createInterface({ input }).on('line', (line) => {
+					// If this was a FROM line, take the last found
+					// image id and save it as a stage id
+					// Source: https://github.com/balena-io/balena-cli/blob/f6d668684a6f5ea8102a964ca1942b242eaa7ae2/lib/utils/device/live.ts#L300-L325
+					if (
+						/step \d+(?:\/\d+)?\s*:\s*FROM/i.test(line) &&
+						lastArrowMessage != null
+					) {
+						ids.push(lastArrowMessage);
+					} else {
+						const msg = extractDockerArrowMessage(line);
+						if (msg != null) {
+							lastArrowMessage = msg;
+						}
+					}
+
+					// Log the build line
+					console.info(line);
+				});
+
+				// stream.pipe(bufStream);
+				tarStream.pipe(input);
 			},
 		});
 	});

sync/init.ts

@@ -14,39 +14,6 @@ interface Opts {
 	arch?: string;
 }
 
-// Source: https://github.com/balena-io/balena-cli/blob/f6d668684a6f5ea8102a964ca1942b242eaa7ae2/lib/utils/device/live.ts#L539-L547
-function extractDockerArrowMessage(outputLine: string): string | undefined {
-	const arrowTest = /^.*\s*-+>\s*(.+)/i;
-	const match = arrowTest.exec(outputLine);
-	if (match != null) {
-		return match[1];
-	}
-}
-
-// Source: https://github.com/balena-io/balena-cli/blob/f6d668684a6f5ea8102a964ca1942b242eaa7ae2/lib/utils/device/live.ts#L300-L325
-function getMultiStateImageIDs(buildLog: string): string[] {
-	const ids = [] as string[];
-	const lines = buildLog.split(/\r?\n/);
-	let lastArrowMessage: string | undefined;
-	for (const line of lines) {
-		// If this was a from line, take the last found
-		// image id and save it
-		if (
-			/step \d+(?:\/\d+)?\s*:\s*FROM/i.test(line) &&
-			lastArrowMessage != null
-		) {
-			ids.push(lastArrowMessage);
-		} else {
-			const msg = extractDockerArrowMessage(line);
-			if (msg != null) {
-				lastArrowMessage = msg;
-			}
-		}
-	}
-
-	return ids;
-}
-
 function getPathPrefix(arch: string) {
 	switch (arch) {
 		/**
@@ -74,7 +41,7 @@ export async function initDevice(opts: Opts) {
 
 	const buildCache = await device.readBuildCache(opts.address);
 
-	const buildLog = await device.performBuild(opts.docker, opts.dockerfile, {
+	const stageImages = await device.performBuild(opts.docker, opts.dockerfile, {
 		buildargs: { ARCH: arch, PREFIX: getPathPrefix(arch) },
 		t: image,
 		labels: { 'io.balena.livepush-image': '1', 'io.balena.architecture': arch },
@@ -84,8 +51,6 @@ export async function initDevice(opts: Opts) {
 		nocache: opts.nocache,
 	});
 
-	const stageImages = getMultiStateImageIDs(buildLog);
-
 	// Store the list of stage images for the next time the sync
 	// command is called. This will only live until the device is rebooted
 	await device.writeBuildCache(opts.address, stageImages);

sync/sync.ts

@@ -93,12 +93,12 @@ const argv = yargs
 			sigint = () => reject(new Error('User interrupt (Ctrl+C) received'));
 			process.on('SIGINT', sigint);
 		});
-	} catch (e) {
+	} catch (e: any) {
 		console.error('Error:', e.message);
 	} finally {
 		console.info('Cleaning up. Please wait ...');
 		await cleanup();
 		process.removeListener('SIGINT', sigint);
-		process.exit(1);
+		process.exit(0);
 	}
 })();