Fix bugs in stream code

Author: fbbdev

lib/streams.js

@@ -1,9 +1,6 @@
 /**
  * Copyright (c) 2016 Fabio Massaioli and other contributors
  *
- * Code from Node http module:
- *   Copyright Joyent, Inc. and other Node contributors
- *
  * Permission is hereby granted, free of charge, to any person obtaining a copy of
  * this software and associated documentation files (the "Software"), to deal in
  * the Software without restriction, including without limitation the rights to
@@ -30,6 +27,7 @@ var stream = require('stream'),
 
 exports.InputStream = InputStream;
 exports.OutputStream = OutputStream;
+exports.IOStream = IOStream;
 
 /**
  * function InputStream([buffer])
@@ -73,9 +71,11 @@ function OutputStream(conn, id, recordType) {
 }
 util.inherits(OutputStream, stream.Writable);
 
-OutputStream.prototype.close = function () {
-    this._open = false;
-    this.emit('close');
+OutputStream.prototype._close = function (hadError) {
+    if (this._open) {
+        this._open = false;
+        this.emit('close', hadError ? true : false);
+    }
 };
 
 OutputStream.prototype._write = function (chunk, encoding, callback) {
@@ -90,6 +90,11 @@ OutputStream.prototype._write = function (chunk, encoding, callback) {
         chunk = new Buffer(chunk, encoding);
     }
 
+    if (chunk.length === 0) {
+        callback.call(this);
+        return;
+    }
+
     if (chunk.length <= 65535) {
         chunks.push(chunk);
     } else {
@@ -102,15 +107,34 @@ OutputStream.prototype._write = function (chunk, encoding, callback) {
     }
 
     while (chunks.length > 1) {
-        this._conn.stream.writeRecord(this._id, new this.type(chunks.shift()));
+        this._conn.stream.writeRecord(
+            this._id, new this.recordType(chunks.shift()));
     }
 
     this._conn.stream.writeRecord(
         this._id,
-        new this.type(chunks.shift()),
+        new this.recordType(chunks.shift()),
         callback.bind(this));
 };
 
+OutputStream.prototype.write = function () {
+    if (!this._open) {
+        this.emit('error', new Error("Output stream is not open"));
+        return;
+    }
+
+    return stream.Writable.prototype.write.apply(this, arguments);
+}
+
+OutputStream.prototype.end = function () {
+    if (!this._open) {
+        this.emit('error', new Error("Output stream is not open"));
+        return;
+    }
+
+    return stream.Writable.prototype.end.apply(this, arguments);
+}
+
 /**
  * function IOStream([buffer])
  * Duplex stream interface for FastCGI input streams
@@ -134,5 +158,7 @@ util.inherits(IOStream, stream.Duplex);
 IOStream.prototype._data = InputStream.prototype._data;
 IOStream.prototype._read = InputStream.prototype._read;
 
-IOStream.prototype.close = OutputStream.prototype.close;
+IOStream.prototype._close = OutputStream.prototype._close;
 IOStream.prototype._write = OutputStream.prototype._write;
+IOStream.prototype.write = OutputStream.prototype.write;
+IOStream.prototype.end = OutputStream.prototype.end;