Adds stream to the request body

examples/simple.js

@@ -20,11 +20,11 @@ builder
 
       //sys.puts("Request: " + requestNumber);
 
-      env.request.addListener("body", function(data){
+      env.request.body.addListener("data", function(data){
         buffer.push(data);
       })
 
-      env.request.addListener("complete", function(){
+      env.request.body.addListener("finish", function(){
        // sys.puts("Complete: " + requestNumber);
         env.response.headers['content-type'] = 'text/html';
         env.response.body = "<h1>Hello jello world</h1>" + buffer.join("");

lib/chain/request.js

@@ -2,6 +2,7 @@ var http    = require('http');
 var events  = require('events');
 var sys     = require('sys');
 var Url     = require('url');
+var Stream  = require('./vendor/stream');
 
 // parts of this based on the request object from http://github.com/kriszyp/jsgi-node
 
@@ -11,12 +12,7 @@ function Request(request, chain){
     url               = Url.parse( request.url ),
     headers           = request.headers,
     namePort          = headers.host.split( ":" ),
-    lowerCaseHeaders  = {},
-    self              = this,
-    bodyListener      = false,
-    inputBuffer       = [],
-    bodyComplete      = false,
-    completeEmitted   = false;
+    lowerCaseHeaders  = {};
 
   this.chain          = chain;
   this.method         = request.method;
@@ -29,6 +25,24 @@ function Request(request, chain){
   this.scheme         = "http";
   this.headers        = request.headers;
   this.Connection     = request.Connection;
+  var body = this.body = new Stream();
+
+  body.pause();
+
+  request.addListener("body", function(chunk){ body.write(chunk) })
+  request.addListener("complete", function() { body.close() })
+
+  body.addListener("newListener", function(type, listener){
+    if(type == "data")
+      body.resume();
+  })
+
+  body.addListener("pause",   function(){
+    sys.puts("Pausing the request");
+    request.pause()
+  });
+  body.addListener("resume",  function(){request.resume()});
+  body.addListener("eof",     function(){body.emit("finish")});
 
   for(var i in headers){
     lowerCaseHeaders[i.toLowerCase()] = headers[i];
@@ -37,47 +51,6 @@ function Request(request, chain){
   this.version = [ request.httpVersionMajor, request.httpVersionMinor ];
 
   request.isComplete  = false;
-
-  this.addListener = function(event){
-    if(event == "body"){
-      if( self.bodyEmitStarted )
-        throw(new Error("Request Body Emitting Already Started"));
-
-      bodyListener = true;
-    }
-    Request.prototype.addListener.apply(this, arguments);
-
-    // Have to give the complete listener a chance to be added
-    process.nextTick(shiftAndEmit)
-  }
-
-  // From here we deal with buffering the request as it comes in
-  function shiftAndEmit(){
-    if(!bodyListener) return;
-
-    if(inputBuffer.length > 0){
-      var chunk = inputBuffer.shift();
-      self.bodyEmitStarted = true;
-      self.emit("body", chunk);
-    }
-
-    if(bodyComplete && !completeEmitted && inputBuffer.length == 0){
-      self.emit("complete");
-      completeEmitted = true;
-    } else if(inputBuffer.length > 0) {
-      process.nextTick(shiftAndEmit);
-    }
-  }
-
-  request
-    .addListener( "body", function( data ) {
-      inputBuffer.push( data );
-      shiftAndEmit();
-    })
-    .addListener("complete", function() {
-      bodyComplete = true;
-      shiftAndEmit();
-    });
 }
 
 sys.inherits(Request, events.EventEmitter);

lib/chain/vendor/stream.js

@@ -0,0 +1,50 @@
+
+module.exports = Stream;
+
+var Emitter = require("events").EventEmitter;
+
+function Stream () {
+  Emitter.call(this);
+
+  var buffer = [];
+  this.pause = function () {
+    this.emit("pause");
+    buffer.paused = true;
+  };
+  this.resume = function () {
+    this.emit("resume");
+    buffer.paused = false;
+    flow(this, buffer);
+  };
+  this.write = function (data) {
+    if (buffer.closed) throw new Error("Cannot write after EOF.");
+    buffer.push(data);
+    flow(this, buffer);
+  };
+  this.close = function () {
+    buffer.closed = true;
+    flow(this, buffer);
+  };
+};
+
+function flow (stream, buffer) {
+  if (buffer.flowing || buffer.paused) return;
+  buffer.flowing = true;
+  process.nextTick(function () {
+    buffer.flowing = false;
+    write(stream, buffer);
+  });
+};
+function write (stream, buffer) {
+  if (buffer.paused) return;
+  if (buffer.length === 0) {
+    stream.emit("drain");
+    if (buffer.closed) stream.emit("eof");
+    return;
+  }
+  var chunk = buffer.shift();
+  stream.emit("data", chunk);
+  flow(stream, buffer);
+};
+
+Stream.prototype.__proto__ = Emitter.prototype;