background - persistence pipeline - fix persistence bug

1 files changed, 24 insertions, 0 deletions

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)
+  }
+
+}