handle stdout backpressure 🐛

cli.js

@@ -40,21 +40,28 @@ if (argv.version || argv.v) {
 }
 
 const {basename, extname} = require('path')
+const {pipeline} = require('stream')
 const convertGtfsToSql = require('./index')
 
 const files = argv._.map((file) => {
 	const name = basename(file, extname(file))
 	return {name, file}
 })
 
-convertGtfsToSql(files, {
+const opt = {
 	silent: !!(argv.silent || argv.s),
 	requireDependencies: !!(argv['require-dependencies'] || argv.d),
 	ignoreUnsupportedFiles: !!(argv['ignore-unsupported'] || argv.u),
 	tripsWithoutShapeId: !!argv['trips-without-shape-id'],
 	routesWithoutAgencyId: !!argv['routes-without-agency-id'],
-})
-.catch((err) => {
-	if (err && err.code !== 'EPIPE') console.error(err)
-	process.exit(1)
-})
+}
+
+pipeline(
+	convertGtfsToSql(files, opt),
+	process.stdout,
+	(err) => {
+		if (!err) return;
+		if (err.code !== 'EPIPE') console.error(err)
+		process.exit(1)
+	}
+)

index.js

@@ -7,7 +7,7 @@ const formatters = require('./lib')
 const getDependencies = require('./lib/deps')
 const pkg = require('./package.json')
 
-const convertGtfsToSql = async (files, opt = {}) => {
+const convertGtfsToSql = async function* (files, opt = {}) {
 	opt = {
 		silent: false,
 		requireDependencies: false,
@@ -48,19 +48,19 @@ const convertGtfsToSql = async (files, opt = {}) => {
 	const order = []
 	sequencify(tasks, files.map(f => f.name), order)
 
-	process.stdout.write(`\
+	yield `\
 -- GTFS SQL dump generated by ${pkg.name} v${pkg.version}
 -- ${pkg.homepage}
 \\set ON_ERROR_STOP True
 CREATE EXTENSION IF NOT EXISTS postgis;
 BEGIN;
-\n`)
+\n`
 
 	const csv = new Stringifier({quoted: true})
 
 	for (const name of order) {
 		if (!silent) console.error(name)
-		process.stdout.write(`-- ${name}\n-----------------\n\n`)
+		yield `-- ${name}\n-----------------\n\n`
 
 		const {file} = tasks[name]
 		const {
@@ -70,26 +70,26 @@ BEGIN;
 		} = formatters[name]
 
 		if ('string' === typeof beforeAll && beforeAll) {
-			process.stdout.write(beforeAll)
+			yield beforeAll
 		} else if ('function' === typeof beforeAll) {
-			process.stdout.write(beforeAll(opt))
+			yield beforeAll(opt)
 		}
 
 		let n = 0
 		for await (const row of readCsv(file)) {
 			const formatted = csv.stringify(formatRow(row, opt))
-			process.stdout.write(formatted + '\n')
+			yield formatted + '\n'
 		}
 
 		if ('string' === typeof afterAll && afterAll) {
-			process.stdout.write(afterAll + ';\n')
+			yield afterAll + ';\n'
 		} else if ('function' === typeof afterAll) {
-			process.stdout.write(afterAll(opt) + ';\n')
+			yield afterAll(opt) + ';\n'
 		}
 	}
 
-	process.stdout.write(`\
-COMMIT;`)
+	yield `\
+COMMIT;`
 }
 
 module.exports = convertGtfsToSql