Refactor streams

Author: fbbdev

lib/streams.js

@@ -33,98 +33,51 @@ exports.OutputStream = OutputStream;
 
 /**
  * function InputStream([buffer])
- * Readable stream interface for buffer arrays
+ * Readable stream interface for FastCGI input streams
  */
 
-function InputStream(buffer) {
-    if (!(this instanceof InputStream)) {
-        return new InputStream(array);
-    }
-
+function InputStream() {
     stream.Readable.call(this);
 
-    this.buffer = buffer || [];
-    this.writable = true;
+    this._buffer = [];
+    this._canPush = true;
 }
 util.inherits(InputStream, stream.Readable);
 
-InputStream.prototype.append = function (chunk) {
-    if (this.writable) {
-        this.writable = this.push(chunk);
+InputStream.prototype._data = function (chunk) {
+    if (this._canPush) {
+        this._canPush = this.push(chunk);
     } else {
-        this.buffer.push(chunk);
+        this._buffer.push(chunk);
     }
 };
 
 InputStream.prototype._read = function (size) {
-    while (this.buffer.length && (this.writable = this.push(this.buffer.shift())));
+    while (this._buffer.length && (this._canPush = this.push(this._buffer.shift())));
 };
 
 /**
- * function OutputStream(conn, req, res, type)
+ * function OutputStream(conn, type)
  * Writable stream interface for FastCGI output streams
  */
 
-function OutputStream(conn, req, res, type) {
-    if (!(this instanceof OutputStream)) {
-        return new OutputStream(conn, req, res, type);
-    }
-
+function OutputStream(conn, id, recordType) {
     stream.Writable.call(this);
 
-    this.conn = conn;
-    this.req = req;
-    this.res = this._httpMessage = res; // NOTE: http.OutgoingMessage needs connection._httpMessage = response
-    this.type = type || fcgi.records.StdOut;
+    this.recordType = recordType || fcgi.records.StdOut;
 
-    this.timeout = 0;
+    this._conn = conn;
+    this._id = id;
 
-    // NOTE: http.OutgoingMessage needs connection.writable = true
-    this._open = this.writable = true;
-    this.on('drain', this.res.emit.bind(res, 'drain'));
+    this._open = true;
 }
 util.inherits(OutputStream, stream.Writable);
 
 OutputStream.prototype.close = function () {
-    this._open = this.writable = false;
+    this._open = false;
     this.emit('close');
 };
 
-OutputStream.prototype.setTimeout = function (msecs, callback) {
-    if (this._timeout_ref) {
-        clearTimeout(this._timeout_ref);
-    }
-
-    this.timeout = msecs;
-    if (callback) {
-        this.once('timeout', callback);
-    }
-
-    if (msecs > 0) {
-        this._timeout_ref = setTimeout(function (self) {
-            self.emit('timeout');
-        }, msecs, this);
-    } else {
-        this._timeout_ref = undefined;
-    }
-};
-
-OutputStream.prototype._resetTimeout = function () {
-    if (this.timeout || this._timeout_ref) {
-        if (this._timeout_ref) {
-            clearTimeout(this._timeout_ref);
-        }
-
-        if (this.timeout > 0) {
-            this._timeout_ref = setTimeout(function (self) {
-                self.emit('timeout');
-            }, this.timeout, this);
-        } else {
-            this._timeout_ref = undefined;
-        }
-    }
-};
-
 OutputStream.prototype._write = function (chunk, encoding, callback) {
     if (!this._open) {
         callback(new Error("Output stream is not open"));
@@ -133,8 +86,6 @@ OutputStream.prototype._write = function (chunk, encoding, callback) {
 
     var chunks = [];
 
-    this._resetTimeout();
-
     if (!Buffer.isBuffer(chunk)) {
         chunk = new Buffer(chunk, encoding);
     }
@@ -151,13 +102,11 @@ OutputStream.prototype._write = function (chunk, encoding, callback) {
     }
 
     while (chunks.length > 1) {
-        this.conn.stream.writeRecord(
-            this.req.id,
-            new this.type(chunks.shift()));
+        this._conn.stream.writeRecord(this._id, new this.type(chunks.shift()));
     }
 
-    this.conn.stream.writeRecord(
-        this.req.id,
+    this._conn.stream.writeRecord(
+        this._id,
         new this.type(chunks.shift()),
         callback.bind(this));
 };