From 7b414f3ed08b8eb35ce7a8e076e4ffd75fea3d30 Mon Sep 17 00:00:00 2001 From: kumavis Date: Wed, 13 Jun 2018 16:45:18 -0700 Subject: background - persistence pipeline - fix persistence bug --- app/scripts/lib/createStreamSink.js | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) create mode 100644 app/scripts/lib/createStreamSink.js (limited to 'app/scripts/lib/createStreamSink.js') diff --git a/app/scripts/lib/createStreamSink.js b/app/scripts/lib/createStreamSink.js new file mode 100644 index 000000000..cf9416fea --- /dev/null +++ b/app/scripts/lib/createStreamSink.js @@ -0,0 +1,24 @@ +const WritableStream = require('readable-stream').Writable +const promiseToCallback = require('promise-to-callback') + +module.exports = createStreamSink + + +function createStreamSink(asyncWriteFn, _opts) { + return new AsyncWritableStream(asyncWriteFn, _opts) +} + +class AsyncWritableStream extends WritableStream { + + constructor (asyncWriteFn, _opts) { + const opts = Object.assign({ objectMode: true }, _opts) + super(opts) + this._asyncWriteFn = asyncWriteFn + } + + // write from incomming stream to state + _write (chunk, encoding, callback) { + promiseToCallback(this._asyncWriteFn(chunk, encoding))(callback) + } + +} -- cgit v1.2.3