stream_wrap: introduce StreamWrapCallbacks

StreamWrapCallbacks is a helper class for incepting into uv_stream_t*
management process.

1 files changed, 78 insertions, 1 deletions

diff --git a/src/stream_wrap.h b/src/stream_wrap.h
index 8b58c8af6..6b73efd68 100644
--- a/src/stream_wrap.h
+++ b/src/stream_wrap.h
@@ -25,18 +25,83 @@
 #include "v8.h"
 #include "node.h"
 #include "handle_wrap.h"
+#include "req_wrap.h"
 #include "string_bytes.h"
 
 namespace node {
 
 // Forward declaration
-class WriteWrap;
+class StreamWrap;
 
+typedef class ReqWrap<uv_shutdown_t> ShutdownWrap;
+
+class WriteWrap: public ReqWrap<uv_write_t> {
+ public:
+  explicit WriteWrap(StreamWrap* wrap) {
+    wrap_ = wrap;
+  }
+
+  void* operator new(size_t size, char* storage) { return storage; }
+
+  // This is just to keep the compiler happy. It should never be called, since
+  // we don't use exceptions in node.
+  void operator delete(void* ptr, char* storage) { assert(0); }
+
+  StreamWrap* wrap_;
+
+ protected:
+  // People should not be using the non-placement new and delete operator on a
+  // WriteWrap. Ensure this never happens.
+  void* operator new(size_t size) { assert(0); };
+  void operator delete(void* ptr) { assert(0); };
+};
+
+// Overridable callbacks' types
+class StreamWrapCallbacks {
+ public:
+  explicit StreamWrapCallbacks(StreamWrap* wrap) : wrap_(wrap) {
+  }
+
+  explicit StreamWrapCallbacks(StreamWrapCallbacks* old) : wrap_(old->wrap_) {
+  }
+
+  virtual ~StreamWrapCallbacks() {
+  }
+
+  virtual int DoWrite(WriteWrap* w,
+                      uv_buf_t* bufs,
+                      size_t count,
+                      uv_stream_t* send_handle,
+                      uv_write_cb cb);
+  virtual void AfterWrite(WriteWrap* w);
+  virtual uv_buf_t DoAlloc(uv_handle_t* handle, size_t suggested_size);
+  virtual void DoRead(uv_stream_t* handle,
+                      ssize_t nread,
+                      uv_buf_t buf,
+                      uv_handle_type pending);
+  virtual int DoShutdown(ShutdownWrap* req_wrap, uv_shutdown_cb cb);
+
+  v8::Handle<v8::Object> Self();
+
+ protected:
+  StreamWrap* wrap_;
+};
 
 class StreamWrap : public HandleWrap {
  public:
   uv_stream_t* GetStream() { return stream_; }
 
+  void OverrideCallbacks(StreamWrapCallbacks* callbacks) {
+    StreamWrapCallbacks* old = callbacks_;
+    callbacks_ = callbacks;
+    if (old != &default_callbacks_)
+      delete old;
+  }
+
+  StreamWrapCallbacks* GetCallbacks() {
+    return callbacks_;
+  }
+
   static void Initialize(v8::Handle<v8::Object> target);
 
   static v8::Handle<v8::Value> GetFD(v8::Local<v8::String>,
@@ -53,10 +118,19 @@ class StreamWrap : public HandleWrap {
   static v8::Handle<v8::Value> WriteUtf8String(const v8::Arguments& args);
   static v8::Handle<v8::Value> WriteUcs2String(const v8::Arguments& args);
 
+  // Overridable callbacks
+  StreamWrapCallbacks* callbacks_;
+
  protected:
   static size_t WriteBuffer(v8::Handle<v8::Value> val, uv_buf_t* buf);
 
   StreamWrap(v8::Handle<v8::Object> object, uv_stream_t* stream);
+  ~StreamWrap() {
+    if (callbacks_ != &default_callbacks_) {
+      delete callbacks_;
+      callbacks_ = NULL;
+    }
+  }
   void StateChange() { }
   void UpdateWriteQueueSize();
 
@@ -79,6 +153,9 @@ class StreamWrap : public HandleWrap {
 
   size_t slab_offset_;
   uv_stream_t* stream_;
+
+  StreamWrapCallbacks default_callbacks_;
+  friend class StreamWrapCallbacks;
 };