Add Request objects on the HTTP server can be interrupted.

deps/http_parser/http_parser.h

@@ -71,7 +71,7 @@ struct http_parser {
 
   size_t chunk_size;
   unsigned eating:1;
-  unsigned buffer_overflow:1;
+  unsigned error:1;
   size_t body_read;
 
   const char *header_field_mark; 

deps/http_parser/http_parser.rl

@@ -48,7 +48,7 @@ do {                                                                 \
   if (parser->FOR##_mark) {                                          \
     parser->FOR##_size += p - parser->FOR##_mark;                    \
     if (parser->FOR##_size > MAX_FIELD_SIZE) {                       \
-      parser->buffer_overflow = TRUE;                                \
+      parser->error = TRUE;                                          \
       return 0;                                                      \
     }                                                                \
     if (parser->on_##FOR) {                                          \
@@ -142,42 +142,60 @@ do {                                                                 \
 
   action header_field {
     CALLBACK(header_field);
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
     parser->header_field_mark = NULL;
     parser->header_field_size = 0;
   }
 
   action header_value {
     CALLBACK(header_value);
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
     parser->header_value_mark = NULL;
     parser->header_value_size = 0;
   }
 
   action request_uri { 
     CALLBACK(uri);
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
     parser->uri_mark = NULL;
     parser->uri_size = 0;
   }
 
   action fragment { 
     CALLBACK(fragment);
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
     parser->fragment_mark = NULL;
     parser->fragment_size = 0;
   }
 
   action query_string { 
     CALLBACK(query_string);
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
     parser->query_string_mark = NULL;
     parser->query_string_size = 0;
   }
 
   action request_path {
     CALLBACK(path);
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
     parser->path_mark = NULL;
     parser->path_size = 0;
   }
@@ -185,20 +203,26 @@ do {                                                                 \
   action headers_complete {
     if(parser->on_headers_complete) {
       callback_return_value = parser->on_headers_complete(parser);
-      if (callback_return_value != 0) fbreak;
+      if (callback_return_value != 0) {
+        parser->error = TRUE;
+        return 0;
+      }
     }
   }
 
   action begin_message {
     if(parser->on_message_begin) {
       callback_return_value = parser->on_message_begin(parser);
-      if (callback_return_value != 0) fbreak;
+      if (callback_return_value != 0) {
+        parser->error = TRUE;
+        return 0;
+      }
     }
   }
 
   action content_length {
     if (parser->content_length > INT_MAX) {
-      parser->buffer_overflow = TRUE;
+      parser->error = TRUE;
       return 0;
     }
     parser->content_length *= 10;
@@ -233,7 +257,10 @@ do {                                                                 \
 
   action skip_chunk_data {
     SKIP_BODY(MIN(parser->chunk_size, REMAINING));
-    if (callback_return_value != 0) fbreak;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
 
     fhold; 
     if (parser->chunk_size > REMAINING) {
@@ -261,7 +288,11 @@ do {                                                                 \
       p += 1;  
 
       SKIP_BODY(MIN(REMAINING, parser->content_length));
-      if (callback_return_value != 0) fbreak;
+
+      if (callback_return_value != 0) {
+        parser->error = TRUE;
+        return 0;
+      }
 
       fhold;
       if(parser->chunk_size > REMAINING) {
@@ -390,7 +421,7 @@ http_parser_init (http_parser *parser, enum http_parser_type type)
   %% write init;
   parser->cs = cs;
   parser->type = type;
-  parser->buffer_overflow = 0;
+  parser->error = 0;
 
   parser->on_message_begin = NULL;
   parser->on_path = NULL;
@@ -420,7 +451,10 @@ http_parser_execute (http_parser *parser, const char *buffer, size_t len)
   if (0 < parser->chunk_size && parser->eating) {
     /* eat body */
     SKIP_BODY(MIN(len, parser->chunk_size));
-    if (callback_return_value != 0) goto out;
+    if (callback_return_value != 0) {
+      parser->error = TRUE;
+      return 0;
+    }
   }
 
   if (parser->header_field_mark)   parser->header_field_mark   = buffer;
@@ -441,7 +475,6 @@ http_parser_execute (http_parser *parser, const char *buffer, size_t len)
   CALLBACK(path);
   CALLBACK(uri);
 
-out:
   assert(p <= pe && "buffer overflow after parsing execute");
   return(p - buffer);
 }
@@ -449,7 +482,7 @@ out:
 int
 http_parser_has_error (http_parser *parser) 
 {
-  if (parser->buffer_overflow) return TRUE;
+  if (parser->error) return TRUE;
   return parser->cs == http_parser_error;
 }
 

src/http.cc

@@ -80,9 +80,7 @@ void
 HTTPConnection::OnReceive (const void *buf, size_t len)
 {
   http_parser_execute(&parser_, static_cast<const char*>(buf), len);
-
-  if (http_parser_has_error(&parser_))
-    ForceClose();
+  if (http_parser_has_error(&parser_)) ForceClose();
 }
 
 int
@@ -187,16 +185,15 @@ HTTPConnection::on_headers_complete (http_parser *parser)
           ); 
   message_handler->Set(HTTP_VERSION_SYMBOL, String::New(version));
 
-  // SHOULD KEEP ALIVE
-  message_handler->Set( SHOULD_KEEP_ALIVE_SYMBOL
-                      , http_parser_should_keep_alive(&connection->parser_) ? True() : False()
-                      );
+  message_handler->Set(SHOULD_KEEP_ALIVE_SYMBOL, 
+      http_parser_should_keep_alive(&connection->parser_) ? True() : False());
 
-
-  Local<Value> on_headers_complete_v = message_handler->Get(ON_HEADERS_COMPLETE_SYMBOL);
+  Local<Value> on_headers_complete_v =
+    message_handler->Get(ON_HEADERS_COMPLETE_SYMBOL);
   if (on_headers_complete_v->IsFunction() == false) return 0;
 
-  Handle<Function> on_headers_complete = Handle<Function>::Cast(on_headers_complete_v);
+  Handle<Function> on_headers_complete =
+    Handle<Function>::Cast(on_headers_complete_v);
 
   TryCatch try_catch;
   Local<Value> ret = on_headers_complete->Call(message_handler, 0, NULL);
@@ -262,12 +259,12 @@ HTTPConnection::on_message_complete (http_parser *parser)
 
   Local<Value> message_handler_v = 
     connection->handle_->GetHiddenValue(MESSAGE_HANDLER_SYMBOL);
-  Local<Object> message_handler = message_handler_v->ToObject();
   connection->handle_->DeleteHiddenValue(MESSAGE_HANDLER_SYMBOL);
 
+  Local<Object> message_handler = message_handler_v->ToObject();
+
   Local<Value> on_msg_complete_v = message_handler->Get(ON_MESSAGE_COMPLETE_SYMBOL);
-  if (on_msg_complete_v->IsFunction() == false) 
-    return 0;
+  if (on_msg_complete_v->IsFunction() == false) return 0;
   Handle<Function> on_msg_complete = Handle<Function>::Cast(on_msg_complete_v);
 
   TryCatch try_catch;
@@ -295,12 +292,6 @@ HTTPConnection::HTTPConnection (Handle<Object> handle, enum http_parser_type typ
   parser_.data = this;
 }
 
-
-HTTPConnection::~HTTPConnection ( )
-{
-}
-
-
 Persistent<FunctionTemplate> HTTPServer::constructor_template;
 
 void

src/http.h

@@ -20,7 +20,6 @@ protected:
 
   HTTPConnection (v8::Handle<v8::Object> handle, 
                   enum http_parser_type type);
-  ~HTTPConnection ( );
 
   void OnReceive (const void *buf, size_t len);
 

src/http.js

@@ -239,22 +239,24 @@ node.http.Server = function (RequestHandler, options) {
     var responses = [];
 
     connection.onMessage = function ( ) {
+      var interrupted = false;
                                             // filled in ...
       var req = { method          : null    // at onHeadersComplete
                 , uri             : ""      // at onURI
-                , httpVersion    : null    // at onHeadersComplete
+                , httpVersion     : null    // at onHeadersComplete
                 , headers         : []      // at onHeaderField, onHeaderValue
                 , onBody          : null    // by user
                 , onBodyComplete  : null    // by user
+                , interrupt       : function ( ) { interrupted = true; }
                 , setBodyEncoding : function (enc) {
                     connection.setEncoding(enc);
                   }
-                }
+                };
       var res = new node.http.ServerResponse(connection, responses);
 
       this.onURI = function (data) {
         req.uri += data;
-        return true
+        return !interrupted;
       };
 
       var last_was_value = false;
@@ -266,7 +268,7 @@ node.http.Server = function (RequestHandler, options) {
         else
           headers.push([data]);
         last_was_value = false;
-        return true;
+        return !interrupted;
       };
 
       this.onHeaderValue = function (data) {
@@ -276,31 +278,29 @@ node.http.Server = function (RequestHandler, options) {
         else 
           last_pair[1] += data;
         last_was_value = true;
-        return true;
+        return !interrupted;
       };
 
       this.onHeadersComplete = function () {
         req.httpVersion = this.httpVersion;
-        req.method       = this.method;
-        req.uri          = node.http.parseUri(req.uri); // TODO parse the URI lazily
-
+        req.method = this.method;
+        // TODO parse the URI lazily?
+        req.uri = node.http.parseUri(req.uri);
         res.should_keep_alive = this.should_keep_alive;
 
-        return RequestHandler.apply(server, [req, res]);
+        RequestHandler.apply(server, [req, res]);
+
+        return !interrupted;
       };
 
       this.onBody = function (chunk) {
-        if (req.onBody)
-          return req.onBody(chunk);
-        else
-          return true;
+        if (req.onBody) req.onBody(chunk);
+        return !interrupted;
       };
 
-      this.onBodyComplete = function () {
-        if (req.onBodyComplete)
-          return req.onBodyComplete(chunk);
-        else
-          return true;
+      this.onMessageComplete = function () {
+        if (req.onBodyComplete) req.onBodyComplete();
+        return !interrupted;
       };
     };
 
@@ -464,6 +464,7 @@ node.http.Client = function (port, host) {
       res.headers = headers;
 
       req.responseHandler(res);
+      return true;
     };
 
     this.onBody = function (chunk) {

test/test-http-server.js

@@ -43,7 +43,7 @@ function onLoad() {
   c.onReceive = function (chunk) {
     server_response += chunk;
 
-    if ( requests_sent == 1) {
+    if (requests_sent == 1) {
       c.send("POST /quit HTTP/1.1\r\n\r\n");
       c.close();
       assertEquals(c.readyState, "readOnly");

test/test-http.js

@@ -19,10 +19,12 @@ function onLoad () {
       this.close();
     }
 
-    res.sendHeader(200, [["Content-Type", "text/plain"]]);
-    res.sendBody("The path was " + req.uri.path);
-    res.finish();
-    responses_sent += 1;
+    req.onBodyComplete = function () {
+      res.sendHeader(200, [["Content-Type", "text/plain"]]);
+      res.sendBody("The path was " + req.uri.path);
+      res.finish();
+      responses_sent += 1;
+    };
   }).listen(PORT);
 
   var client = new node.http.Client(PORT);

website/api.html

@@ -622,7 +622,7 @@ server.listen(7000, "localhost");</pre>
       <code>"1.1"</code>, <code>"1.0"</code>
     </dd>
     
-    <dt><code>req.onBody</code></dt>
+    <dt><code>req.onBody = function (chunk) { }; </code></dt>
     <dd>
       Callback. Should be set by the user to be informed of when a
       piece of the message body is received. Example:
@@ -640,7 +640,7 @@ req.onBody = function (chunk) {
       </p>
     </dd>
     
-    <dt><code>req.onBodyComplete</code></dt>
+    <dt><code>req.onBodyComplete = function () { };</code></dt>
     <dd>
       Callback. Made exactly once for each message. No arguments.
       After <code>onBodyComplete</code> is executed
@@ -652,6 +652,14 @@ req.onBody = function (chunk) {
       Set the encoding for the request body. Either <code>"utf8"</code>
       or <code>"raw"</code>. Defaults to raw.
     </dd>
+
+    <dt><code>req.interrupt()</code></dt>
+    <dd>
+      Interrupt the request. You will not receive anymore callbacks.
+      This is useful if, for example someone is streaming up a file but it
+      is too large and neesd to be stopped. The connection to the client
+      will be closed immediately. 
+    </dd>
   </dl>
 
   <h3 id="http_server_response"><code>node.http.ServerResponse</code></h3>

website/index.html

@@ -141,7 +141,7 @@ Server running at http://127.0.0.1:8000/</pre>
         being used.  Processes are necessary to scale to multi-core
         computers, not memory-sharing threads. The fundamentals of scalable
         systems are fast networking and non-blocking design&mdash;the rest
-        is message passing.  In the future, I'd like Node to to be able to
+        is message passing.  In the future, I'd like Node to be able to
         spawn new processes (probably using the
         <a href="http://www.whatwg.org/specs/web-workers/current-work/">
           Web Workers API