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 From a8f745f9fe74751b87f500af3857b66d4c80f45e Mon Sep 17 00:00:00 2001 From: brunobar79 Date: Mon, 2 Jul 2018 18:49:33 -0400 Subject: eslint --fix . --- app/scripts/lib/createStreamSink.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'app/scripts/lib/createStreamSink.js') diff --git a/app/scripts/lib/createStreamSink.js b/app/scripts/lib/createStreamSink.js index cf9416fea..b93dbc089 100644 --- a/app/scripts/lib/createStreamSink.js +++ b/app/scripts/lib/createStreamSink.js @@ -4,7 +4,7 @@ const promiseToCallback = require('promise-to-callback') module.exports = createStreamSink -function createStreamSink(asyncWriteFn, _opts) { +function createStreamSink (asyncWriteFn, _opts) { return new AsyncWritableStream(asyncWriteFn, _opts) } -- cgit v1.2.3