'use strict'; const {pipeline: pipelineCb, Writable, Readable} = require('stream'); const util = require('util'); const pipeline = util.promisify(pipelineCb); (async () => { // Create a dummy readable stream let i = 0; const readable = new Readable({ objectMode: true, read () { this.push({counter: i++}); } }); // Create a dummy writable stream with backpressure const writable = new Writable({ objectMode: true, write: (chunk, encoding, callback) => { setTimeout(callback, 10); // Backpressure } }); // Pipeline the sql stream in the dummy writable stream & destroy the writable stream after a few ms try { await Promise.all([ pipeline(readable, writable), (async () => { // Destroy the stream after a few element // pipeline will destroy all the streams await new Promise(resolve => setTimeout(resolve, 50)); writable.destroy(); })() ]); } catch (e) { if (e.code !== 'ERR_STREAM_PREMATURE_CLOSE') { throw e; } } })().catch(e => { console.error(e); process.exitCode = 1; });