Add connection.readPause() and connection.readResume()

src/net.cc

@@ -64,6 +64,8 @@ Connection::Initialize (v8::Handle<v8::Object> target)
   NODE_SET_PROTOTYPE_METHOD(constructor_template, "fullClose", FullClose);
   NODE_SET_PROTOTYPE_METHOD(constructor_template, "forceClose", ForceClose);
   NODE_SET_PROTOTYPE_METHOD(constructor_template, "setEncoding", SetEncoding);
+  NODE_SET_PROTOTYPE_METHOD(constructor_template, "readPause", ReadPause);
+  NODE_SET_PROTOTYPE_METHOD(constructor_template, "readResume", ReadResume);
 
   constructor_template->PrototypeTemplate()->SetAccessor(
       READY_STATE_SYMBOL,
@@ -325,6 +327,32 @@ Connection::SetEncoding (const Arguments& args)
   }
 }
 
+Handle<Value>
+Connection::ReadPause (const Arguments& args)
+{
+  HandleScope scope;
+
+  Connection *connection = ObjectWrap::Unwrap<Connection>(args.This());
+  assert(connection);
+
+  connection->ReadPause();
+
+  return Undefined();
+}
+
+Handle<Value>
+Connection::ReadResume (const Arguments& args)
+{
+  HandleScope scope;
+
+  Connection *connection = ObjectWrap::Unwrap<Connection>(args.This());
+  assert(connection);
+
+  connection->ReadResume();
+
+  return Undefined();
+}
+
 Handle<Value>
 Connection::Close (const Arguments& args)
 {

src/net.h

@@ -27,6 +27,8 @@ protected:
   static v8::Handle<v8::Value> FullClose (const v8::Arguments& args);
   static v8::Handle<v8::Value> ForceClose (const v8::Arguments& args);
   static v8::Handle<v8::Value> SetEncoding (const v8::Arguments& args);
+  static v8::Handle<v8::Value> ReadPause (const v8::Arguments& args);
+  static v8::Handle<v8::Value> ReadResume (const v8::Arguments& args);
 
   static v8::Handle<v8::Value> ReadyStateGetter (v8::Local<v8::String> _,
       const v8::AccessorInfo& info);
@@ -49,6 +51,8 @@ protected:
   void Close (void) { evcom_stream_close(&stream_); }
   void FullClose (void) { evcom_stream_full_close(&stream_); }
   void ForceClose (void) { evcom_stream_force_close(&stream_); }
+  void ReadPause (void) { evcom_stream_read_pause(&stream_); }
+  void ReadResume (void) { evcom_stream_read_resume(&stream_); }
 
   virtual void OnConnect (void);
   virtual void OnReceive (const void *buf, size_t len);

test/mjsunit/test-tcp-throttle.js

@@ -0,0 +1,52 @@
+include("mjsunit.js");
+PORT = 20443;
+N = 500;
+
+server = node.tcp.createServer(function (connection) {
+  function send (j) {
+    if (j >= N) { 
+      connection.fullClose();
+      return;
+    }
+    setTimeout(function () {
+      connection.send("C");
+      send(j+1);
+    }, 10);
+  }
+  send(0);
+});
+server.listen(PORT);
+
+
+recv = "";
+chars_recved = 0;
+
+function onLoad () {
+  client = node.tcp.createConnection(PORT);
+  client.setEncoding("ascii");
+  client.addListener("receive", function (d) {
+      print(d);
+      recv += d;
+  });
+
+  setTimeout(function () {
+    chars_recved = recv.length; 
+    puts("chars_recved: " + chars_recved);
+    assertTrue(chars_recved > 1);
+    client.readPause();
+    setTimeout(function () {
+      puts("chars_recved: " + chars_recved);
+      assertEquals(chars_recved, recv.length);
+      client.readResume();
+    }, 500);
+  }, 100);
+
+  client.addListener("eof", function () {
+    server.close();
+    client.close();
+  });
+}
+
+function onExit () {
+  assertEquals(N, recv.length);
+}

website/api.txt

@@ -1007,7 +1007,6 @@ Either +"closed"+, +"open"+, +"opening"+, +"readOnly"+, or +"writeOnly"+.
 +connection.setEncoding(encoding)+::
 Sets the encoding (either +"utf8"+ or +"raw"+) for data that is received. 
 
-
 +connection.send(data, encoding="ascii")+::
 Sends data on the connection. The data should be eithre an array
 of integers (for raw binary) or a string (for utf8 or ascii).
@@ -1032,6 +1031,13 @@ know about this, just use +close()+.
 Ensures that no more I/O activity happens on this socket. Only
 necessary in case of errors (parse error or so).
 
++connection.readPause()+::
+Pauses the reading of data. That is, +"receive"+ events will not be emitted.
+Useful to throttle back an upload.
+
++connection.readResume()+::
+Resumes reading if reading was paused by +readPause()+.
+
 
 
 === DNS