From 94e87217716ad0dff7061591aaf2149b3b6011dd Mon Sep 17 00:00:00 2001
From: Ryan <ry@tinyclouds.org>
Date: Sun, 9 Aug 2009 18:49:51 +0200
Subject: [PATCH] Add connection.readPause() and connection.readResume()

---
 src/net.cc                        | 28 +++++++++++++++++
 src/net.h                         |  4 +++
 test/mjsunit/test-tcp-throttle.js | 52 +++++++++++++++++++++++++++++++
 website/api.txt                   |  8 ++++-
 4 files changed, 91 insertions(+), 1 deletion(-)
 create mode 100644 test/mjsunit/test-tcp-throttle.js

diff --git a/src/net.cc b/src/net.cc
index 467d8a0591..fcdd8361be 100644
--- a/src/net.cc
+++ b/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)
 {
diff --git a/src/net.h b/src/net.h
index 3b142b2048..25f5e6fb8f 100644
--- a/src/net.h
+++ b/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);
diff --git a/test/mjsunit/test-tcp-throttle.js b/test/mjsunit/test-tcp-throttle.js
new file mode 100644
index 0000000000..51eb60a822
--- /dev/null
+++ b/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);
+}
diff --git a/website/api.txt b/website/api.txt
index 281a511423..29495164c9 100644
--- a/website/api.txt
+++ b/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