Streams: Add _flush capability to writable streams

Author: evansolomon

lib/_stream_transform.js

@@ -126,12 +126,7 @@ function Transform(options) {
   this._readableState.sync = false;
 
   this.once('prefinish', function() {
-    if (util.isFunction(this._flush))
-      this._flush(function(er) {
-        done(stream, er);
-      });
-    else
-      done(stream);
+    done(stream);
   });
 }
 

lib/_stream_writable.js

@@ -434,7 +434,14 @@ function needFinish(stream, state) {
 function prefinish(stream, state) {
   if (!state.prefinished) {
     state.prefinished = true;
-    stream.emit('prefinish');
+    if (util.isFunction(stream._flush)) {
+      stream._flush(function (err) {
+        if (err) return stream.emit('error', err)
+        stream.emit('prefinish')
+      })
+    } else {
+      stream.emit('prefinish');
+    }
   }
 }
 

test/simple/test-stream2-writable.js

@@ -403,3 +403,25 @@ test('finish is emitted if last chunk is empty', function(t) {
   w.write(Buffer(1));
   w.end(Buffer(0));
 });
+
+test('write flush', function(t) {
+  var w = new W();
+  var writtenData = [];
+  w._write = function(chunk, e, cb){
+    writtenData.push(chunk.toString());
+    cb();
+  }
+  w._flush = function(cb){
+    writtenData.push('flushed');
+    cb();
+  }
+
+  w.write('start');
+  w.end();
+
+  w.on('finish', function() {
+    t.equal(writtenData[0], 'start');
+    t.equal(writtenData[1], 'flushed');
+    t.end();
+  });
+});