Implement Promises for file i/o

src/events.cc

@@ -66,6 +66,7 @@ Promise::Initialize (v8::Handle<v8::Object> target)
   Local<FunctionTemplate> t = FunctionTemplate::New();
   constructor_template = Persistent<FunctionTemplate>::New(t);
   constructor_template->Inherit(EventEmitter::constructor_template);
+  constructor_template->InstanceTemplate()->SetInternalFieldCount(1);
 
   // All prototype methods are defined in events.js
 
@@ -80,9 +81,34 @@ Promise::Create (void)
 
   Local<Object> handle =
     Promise::constructor_template->GetFunction()->NewInstance();
+
   Promise *promise = new Promise(handle);
   ObjectWrap::InformV8ofAllocation(promise);
 
+  promise->Attach();
+  ev_unref(EV_DEFAULT_UC);
+
   return promise;
 }
 
+bool
+Promise::EmitSuccess (int argc, v8::Handle<v8::Value> argv[])
+{
+  bool r = Emit("Success", argc, argv);
+
+  Detach();
+  ev_ref(EV_DEFAULT_UC);
+
+  return r;
+}
+
+bool
+Promise::EmitError (int argc, v8::Handle<v8::Value> argv[])
+{
+  bool r = Emit("Error", argc, argv);
+
+  Detach();
+  ev_ref(EV_DEFAULT_UC);
+
+  return r;
+}

src/events.h

@@ -14,6 +14,7 @@ class EventEmitter : public ObjectWrap {
 
   bool Emit (const char *type, int argc, v8::Handle<v8::Value> argv[]);
 
+ protected:
   EventEmitter (v8::Handle<v8::Object> handle)
     : ObjectWrap(handle) { }
 };
@@ -24,18 +25,13 @@ class Promise : public EventEmitter {
   static v8::Persistent<v8::FunctionTemplate> constructor_template;
   virtual size_t size (void) { return sizeof(Promise); };
  
-  bool EmitSuccess (int argc, v8::Handle<v8::Value> argv[])
-  {
-    return Emit("Success", argc, argv);
-  }
-
-  bool EmitError (int argc, v8::Handle<v8::Value> argv[])
-  {
-    return Emit("Error", argc, argv);
-  }
+  bool EmitSuccess (int argc, v8::Handle<v8::Value> argv[]);
+  bool EmitError (int argc, v8::Handle<v8::Value> argv[]);
 
   static Promise* Create (void);
 
+  v8::Handle<v8::Object> Handle(void) { return handle_; }
+
  protected:
 
   Promise (v8::Handle<v8::Object> handle) : EventEmitter(handle) { }

src/events.js

@@ -25,8 +25,13 @@ emitter.emit = function (type, args) {
   if (!this._events.hasOwnProperty(type)) return;
   var listeners = this._events[type];
   var length = listeners.length;
+
   for (var i = 0; i < length; i++) {
-    listeners[i].apply(this, args);
+    if (args) {
+      listeners[i].apply(this, args);
+    } else  {
+      listeners[i].call(this);
+    }
   }
 };
 
@@ -35,10 +40,20 @@ var promise = node.Promise.prototype;
 
 promise.addCallback = function (listener) {
   this.addListener("Success", listener);
+  return this;
 };
 
 promise.addErrback = function (listener) {
   this.addListener("Error", listener);
+  return this;
+};
+
+promise.emitSuccess = function (args) {
+  this.emit("Success", args);
+};
+
+promise.emitError = function (args) {
+  this.emit("Error", args);
 };
 
 })(); // end anonymous namespace

src/file.cc

@@ -1,5 +1,6 @@
 #include "node.h"
 #include "file.h"
+#include "events.h"
 #include <string.h>
 
 #include <sys/types.h>
@@ -27,52 +28,44 @@ using namespace node;
 #define CTIME_SYMBOL       String::NewSymbol("ctime")
 #define BAD_ARGUMENTS      String::New("Bad argument")
 
-#define MAKE_CALLBACK_PTR                                             \
-  Persistent<Function> *callback = NULL;                              \
-  Local<Value> last_arg = args[args.Length()-1];                      \
-  if (last_arg->IsFunction()) {                                       \
-    Local<Function> l = Local<Function>::Cast(last_arg);              \
-    callback = new Persistent<Function>();                            \
-    *callback = Persistent<Function>::New(l);                         \
-  }                                                                   \
-  ev_ref(EV_DEFAULT_UC);
-
-#define CALL_CALLBACK_PTR(req, argc, argv)                            \
-do {                                                                  \
-  if (req->data) {                                                    \
-    Persistent<Function> *callback =                                  \
-      reinterpret_cast<Persistent<Function>*>(req->data);             \
-    TryCatch try_catch;                                               \
-    (*callback)->Call(Context::GetCurrent()->Global(), argc, argv);   \
-    if(try_catch.HasCaught())                                         \
-      node::FatalException(try_catch);                                \
-    delete callback;                                                  \
-  }                                                                   \
-  ev_unref(EV_DEFAULT_UC);                                            \
-} while(0)
-
-#define DEFINE_SIMPLE_CB(name)                                        \
-static int After##name (eio_req *req)                                 \
-{                                                                     \
-  HandleScope scope;                                                  \
-  Local<Value> argv[] = { Integer::New(req->errorno) };               \
-  CALL_CALLBACK_PTR(req, 1, argv);                                    \
-  return 0;                                                           \
-}                                                                     \
-
-DEFINE_SIMPLE_CB(Close)
-static Handle<Value> Close (const Arguments& args)
+static int
+AfterClose (eio_req *req)
+{
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+  if (req->result == 0) {
+    promise->EmitSuccess(0, NULL);
+  } else {
+    promise->EmitError(0, NULL);
+  }
+  return 0;
+}
+
+static Handle<Value>
+Close (const Arguments& args)
 {
   if (args.Length() < 1 || !args[0]->IsInt32())
     return ThrowException(BAD_ARGUMENTS);
   HandleScope scope;
   int fd = args[0]->Int32Value();
-  MAKE_CALLBACK_PTR
-  eio_close(fd, EIO_PRI_DEFAULT, AfterClose, callback);
-  return Undefined();
+
+  Promise *promise = Promise::Create();
+
+  eio_close(fd, EIO_PRI_DEFAULT, AfterClose, promise);
+  return scope.Close(promise->Handle());
+}
+
+static int
+AfterRename (eio_req *req)
+{
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+  if (req->result == 0) {
+    promise->EmitSuccess(0, NULL);
+  } else {
+    promise->EmitError(0, NULL);
+  }
+  return 0;
 }
 
-DEFINE_SIMPLE_CB(Rename)
 static Handle<Value> Rename (const Arguments& args)
 {
   if (args.Length() < 2 || !args[0]->IsString() || !args[1]->IsString())
@@ -80,44 +73,72 @@ static Handle<Value> Rename (const Arguments& args)
   HandleScope scope;
   String::Utf8Value path(args[0]->ToString());
   String::Utf8Value new_path(args[1]->ToString());
-  MAKE_CALLBACK_PTR
-  eio_rename(*path, *new_path, EIO_PRI_DEFAULT, AfterRename, callback);
-  return Undefined();
+
+  Promise *promise = Promise::Create();
+
+  eio_rename(*path, *new_path, EIO_PRI_DEFAULT, AfterRename, promise);
+  return scope.Close(promise->Handle());
+}
+
+static int
+AfterUnlink (eio_req *req)
+{
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+  if (req->result == 0) {
+    promise->EmitSuccess(0, NULL);
+  } else {
+    promise->EmitError(0, NULL);
+  }
+  return 0;
 }
 
-DEFINE_SIMPLE_CB(Unlink)
 static Handle<Value> Unlink (const Arguments& args)
 {
   if (args.Length() < 1 || !args[0]->IsString())
     return ThrowException(BAD_ARGUMENTS);
   HandleScope scope;
   String::Utf8Value path(args[0]->ToString());
-  MAKE_CALLBACK_PTR
-  eio_unlink(*path, EIO_PRI_DEFAULT, AfterUnlink, callback);
-  return Undefined();
+  Promise *promise = Promise::Create();
+  eio_unlink(*path, EIO_PRI_DEFAULT, AfterUnlink, promise);
+  return scope.Close(promise->Handle());
+}
+
+static int
+AfterRMDir (eio_req *req)
+{
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+  if (req->result == 0) {
+    promise->EmitSuccess(0, NULL);
+  } else {
+    promise->EmitError(0, NULL);
+  }
+  return 0;
 }
 
-DEFINE_SIMPLE_CB(RMDir)
 static Handle<Value> RMDir (const Arguments& args)
 {
   if (args.Length() < 1 || !args[0]->IsString())
     return ThrowException(BAD_ARGUMENTS);
   HandleScope scope;
   String::Utf8Value path(args[0]->ToString());
-  MAKE_CALLBACK_PTR
-  eio_rmdir(*path, EIO_PRI_DEFAULT, AfterRMDir, callback);
-  return Undefined();
+  Promise *promise = Promise::Create();
+  eio_rmdir(*path, EIO_PRI_DEFAULT, AfterRMDir, promise);
+  return scope.Close(promise->Handle());
 }
 
 static int
 AfterOpen (eio_req *req)
 {
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+
+  if (req->result < 0) {
+    promise->EmitError(0, NULL);
+    return 0;
+  }
+
   HandleScope scope;
-  const int argc = 2;
-  Local<Value> argv[argc];
-  argv[0] = Integer::New(req->errorno);
-  argv[1] = Integer::New(req->result);
-  CALL_CALLBACK_PTR(req, argc, argv);
+  Local<Value> argv[1] = { Integer::New(req->result) };
+  promise->EmitSuccess(2, argv);
   return 0;
 }
 
@@ -135,27 +156,31 @@ Open (const Arguments& args)
   int flags = args[1]->Int32Value();
   mode_t mode = static_cast<mode_t>(args[2]->Int32Value());
 
-  MAKE_CALLBACK_PTR
+  Promise *promise = Promise::Create();
 
-  eio_open(*path, flags, mode, EIO_PRI_DEFAULT, AfterOpen, callback);
-  return Undefined();
+  eio_open(*path, flags, mode, EIO_PRI_DEFAULT, AfterOpen, promise);
+  return scope.Close(promise->Handle());
 }
 
 static int
 AfterWrite (eio_req *req)
 {
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+
+  if (req->result < 0) {
+    promise->EmitError(0, NULL);
+    return 0;
+  }
+
   HandleScope scope;
 
   free(req->ptr2);
 
   ssize_t written = req->result;
+  Local<Value> argv[1];
+  argv[0] = written >= 0 ? Integer::New(written) : Integer::New(0);
 
-  const int argc = 2;
-  Local<Value> argv[argc];
-  argv[0] = Integer::New(req->errorno);
-  argv[1] = written >= 0 ? Integer::New(written) : Integer::New(0);
-
-  CALL_CALLBACK_PTR(req, argc, argv);
+  promise->EmitSuccess(1, argv);
   return 0;
 }
 
@@ -206,57 +231,63 @@ Write (const Arguments& args)
     return ThrowException(BAD_ARGUMENTS);
   }
 
-  MAKE_CALLBACK_PTR
-  eio_write(fd, buf, len, pos, EIO_PRI_DEFAULT, AfterWrite, callback);
-  return Undefined();
+  Promise *promise = Promise::Create();
+  eio_write(fd, buf, len, pos, EIO_PRI_DEFAULT, AfterWrite, promise);
+  return scope.Close(promise->Handle());
 }
 
 static int
 AfterUtf8Read (eio_req *req)
 {
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+
+  if (req->result < 0) {
+    promise->EmitError(0, NULL);
+    return 0;
+  }
+
   HandleScope scope;
 
-  const int argc = 2;
-  Local<Value> argv[argc];
-  argv[0] = Integer::New(req->errorno);
+  Local<Value> argv[1];
 
-  char *buf = reinterpret_cast<char*>(req->ptr2);
   if (req->result == 0) { 
     // eof 
-    argv[1] = Local<Value>::New(Null());
+    argv[0] = Local<Value>::New(Null());
   } else {
-    argv[1] = String::New(buf, req->result);
+    char *buf = reinterpret_cast<char*>(req->ptr2);
+    argv[0] = String::New(buf, req->result);
   }
 
-  CALL_CALLBACK_PTR(req, argc, argv);
+  promise->EmitSuccess(1, argv);
   return 0;
 }
 
 static int
 AfterRawRead(eio_req *req)
 {
-  HandleScope scope;
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
 
-  const int argc = 2;
-  Local<Value> argv[argc];
-  argv[0] = Integer::New(req->errorno);
+  if (req->result < 0) {
+    promise->EmitError(0, NULL);
+    return 0;
+  }
 
-  char *buf = reinterpret_cast<char*>(req->ptr2);
+  HandleScope scope;
+  Local<Value> argv[1];
 
-  if (req->result == 0) { 
-    // eof 
-    argv[1] = Local<Value>::New(Null());
+  if (req->result == 0) {
+    argv[0] = Local<Value>::New(Null());
   } else {
-    // raw encoding
+    char *buf = reinterpret_cast<char*>(req->ptr2);
     size_t len = req->result;
     Local<Array> array = Array::New(len);
     for (unsigned int i = 0; i < len; i++) {
       array->Set(Integer::New(i), Integer::New(buf[i]));
     }
-    argv[1] = array;
+    argv[0] = array;
   }
 
-  CALL_CALLBACK_PTR(req, argc, argv);
+  promise->EmitSuccess(1, argv);
   return 0;
 }
 
@@ -274,7 +305,7 @@ AfterRawRead(eio_req *req)
 static Handle<Value>
 Read (const Arguments& args)
 {
-  if ( args.Length() < 3 
+  if ( args.Length() < 2 
     || !args[0]->IsInt32()   // fd
     || !args[1]->IsNumber()  // len
      ) return ThrowException(BAD_ARGUMENTS);
@@ -290,56 +321,62 @@ Read (const Arguments& args)
     encoding = static_cast<enum encoding>(args[3]->Int32Value());
   }
 
-  MAKE_CALLBACK_PTR
+  Promise *promise = Promise::Create();
+
   // NOTE: 2nd param: NULL pointer tells eio to allocate it itself
   eio_read(fd, NULL, len, pos, EIO_PRI_DEFAULT, 
-      encoding == UTF8 ? AfterUtf8Read : AfterRawRead, callback);
-  return Undefined();
+      encoding == UTF8 ? AfterUtf8Read : AfterRawRead, promise);
+
+  return scope.Close(promise->Handle());
 }
 
 static int
 AfterStat (eio_req *req)
 {
-  HandleScope scope;
+  Promise *promise = reinterpret_cast<Promise*>(req->data);
+
+  if (req->result < 0) {
+    promise->EmitError(0, NULL);
+    return 0;
+  }
 
-  const int argc = 2;
-  Local<Value> argv[argc];
-  argv[0] = Integer::New(req->errorno);
+  HandleScope scope;
 
+  Local<Value> argv[1];
   Local<Object> stats = Object::New();
   argv[1] = stats;
 
-  if (req->result == 0) {
-    struct stat *s = reinterpret_cast<struct stat*>(req->ptr2);
-
-    /* ID of device containing file */
-    stats->Set(DEV_SYMBOL, Integer::New(s->st_dev));
-    /* inode number */
-    stats->Set(INO_SYMBOL, Integer::New(s->st_ino));
-    /* protection */
-    stats->Set(MODE_SYMBOL, Integer::New(s->st_mode));
-    /* number of hard links */
-    stats->Set(NLINK_SYMBOL, Integer::New(s->st_nlink));
-    /* user ID of owner */
-    stats->Set(UID_SYMBOL, Integer::New(s->st_uid));
-    /* group ID of owner */
-    stats->Set(GID_SYMBOL, Integer::New(s->st_gid));
-    /* device ID (if special file) */
-    stats->Set(RDEV_SYMBOL, Integer::New(s->st_rdev));
-    /* total size, in bytes */
-    stats->Set(SIZE_SYMBOL, Integer::New(s->st_size));
-    /* blocksize for filesystem I/O */
-    stats->Set(BLKSIZE_SYMBOL, Integer::New(s->st_blksize));
-    /* number of blocks allocated */
-    stats->Set(BLOCKS_SYMBOL, Integer::New(s->st_blocks));
-    /* time of last access */
-    stats->Set(ATIME_SYMBOL, Date::New(1000*static_cast<double>(s->st_atime)));
-    /* time of last modification */
-    stats->Set(MTIME_SYMBOL, Date::New(1000*static_cast<double>(s->st_mtime)));
-    /* time of last status change */
-    stats->Set(CTIME_SYMBOL, Date::New(1000*static_cast<double>(s->st_ctime)));
-  }
-  CALL_CALLBACK_PTR(req, argc, argv);                                    \
+  struct stat *s = reinterpret_cast<struct stat*>(req->ptr2);
+
+  /* ID of device containing file */
+  stats->Set(DEV_SYMBOL, Integer::New(s->st_dev));
+  /* inode number */
+  stats->Set(INO_SYMBOL, Integer::New(s->st_ino));
+  /* protection */
+  stats->Set(MODE_SYMBOL, Integer::New(s->st_mode));
+  /* number of hard links */
+  stats->Set(NLINK_SYMBOL, Integer::New(s->st_nlink));
+  /* user ID of owner */
+  stats->Set(UID_SYMBOL, Integer::New(s->st_uid));
+  /* group ID of owner */
+  stats->Set(GID_SYMBOL, Integer::New(s->st_gid));
+  /* device ID (if special file) */
+  stats->Set(RDEV_SYMBOL, Integer::New(s->st_rdev));
+  /* total size, in bytes */
+  stats->Set(SIZE_SYMBOL, Integer::New(s->st_size));
+  /* blocksize for filesystem I/O */
+  stats->Set(BLKSIZE_SYMBOL, Integer::New(s->st_blksize));
+  /* number of blocks allocated */
+  stats->Set(BLOCKS_SYMBOL, Integer::New(s->st_blocks));
+  /* time of last access */
+  stats->Set(ATIME_SYMBOL, Date::New(1000*static_cast<double>(s->st_atime)));
+  /* time of last modification */
+  stats->Set(MTIME_SYMBOL, Date::New(1000*static_cast<double>(s->st_mtime)));
+  /* time of last status change */
+  stats->Set(CTIME_SYMBOL, Date::New(1000*static_cast<double>(s->st_ctime)));
+
+  promise->EmitSuccess(1, argv);
+
   return 0;
 }
 
@@ -353,11 +390,11 @@ Stat (const Arguments& args)
 
   String::Utf8Value path(args[0]->ToString());
 
-  MAKE_CALLBACK_PTR
+  Promise *promise = Promise::Create();
 
-  eio_stat(*path, EIO_PRI_DEFAULT, AfterStat, callback);
+  eio_stat(*path, EIO_PRI_DEFAULT, AfterStat, promise);
 
-  return Undefined();
+  return scope.Close(promise->Handle());
 }
 
 static Handle<Value>

src/file.js

@@ -1,43 +1,49 @@
 node.fs.exists = function (path, callback) {
-  node.fs.stat(path, function (status) {
-    callback(status == 0);
-  });
+  var p = node.fs.stat(path);
+  p.addCallback(function () { callback(true); });
+  p.addErrback(function () { callback(false); });
 }
 
 node.fs.cat = function (path, encoding, callback) {
-  var file = new node.fs.File({encoding: encoding});
-
-  file.onError = function (method, errno, msg) {
-    //node.debug("cat error");
-    callback(-1);
-  };
-
-  var content = (encoding == node.UTF8 ? "" : []);
-  var pos = 0;
-  var chunkSize = 16*1024;
-
-  function readChunk () {
-    file.read(chunkSize, pos, function (chunk) {
-      if (chunk) {
-        if (chunk.constructor == String)
-          content += chunk;
-        else
-          content = content.concat(chunk);
-
-        pos += chunk.length;
-        readChunk();
-      } else {
-        callback(0, content);
-        file.close();
-      }
-    });
-  }
-
-  file.open(path, "r", function () { readChunk(); });
+  var open_promise = node.fs.open(path, node.O_RDONLY, 0666);
+  var cat_promise = new node.Promise();
+
+  encoding = (encoding === "raw" ? node.RAW : node.UTF8);
+
+  open_promise.addErrback(function () { cat_promise.emitError(); });
+  open_promise.addCallback(function (fd) {
+    var content = (encoding == node.UTF8 ? "" : []);
+    var pos = 0;
+
+    function readChunk () {
+      var read_promise = node.fs.read(fd, 16*1024, pos, encoding);
+
+      read_promise.addErrback(function () { cat_promise.emitError(); });
+
+      read_promise.addCallback(function (chunk) {
+        if (chunk) {
+          if (chunk.constructor == String)
+            content += chunk;
+          else
+            content = content.concat(chunk);
+
+          pos += chunk.length;
+          readChunk();
+        } else {
+          cat_promise.emitSuccess([content]);
+          node.fs.close(fd);
+        }
+      });
+    }
+    readChunk();
+  });
+  return cat_promise;
 };
 
 node.fs.File = function (options) {
   var self = this;
+  self.__proto__ = new node.EventEmitter();
+
   options = options || {};
 
   if (options.encoding === "utf8") {
@@ -45,58 +51,65 @@ node.fs.File = function (options) {
   } else {
     self.encoding = node.RAW;
   }
-
   //node.debug("encoding: opts=" + options.encoding + " self=" + self.encoding);
   self.fd = options.fd || null;
 
   var actionQueue = [];
   
   // Adds a method to the queue. 
-  function addAction (method, args, callback) {
-    var action = { method: method 
-                 , callback: callback
-                 , args: args
-                 };
+  function createAction (method, args) {
+    var promise = new node.Promise();
+   
+    promise.method = method;
+    promise.args = args;
+
     //node.debug("add action: " + JSON.stringify(action));
-    actionQueue.push(action);
+    actionQueue.push(promise);
 
     // If the queue was empty, immediately call the method.
     if (actionQueue.length == 1) act();
+
+    return promise;
   }
 
-  // called after each action finishes (when it returns from the thread pool)
-  function poll () {
-    var action = actionQueue[0];
+  function act () {
+    var promise = actionQueue[0]; // peek at the head of the queue
+    if (promise) {
+      node.debug("internal apply " + JSON.stringify(promise.args));
+      internal_methods[promise.method].apply(self, promise.args);
+    }
+  }
 
-    var errno = arguments[0]; 
+  // called after each action finishes (when it returns from the thread pool)
+  function success () {
+    var promise = actionQueue[0];
 
-    //node.debug("poll errno: " + JSON.stringify(errno));
-    //node.debug("poll action: " + JSON.stringify(action));
-    //node.debug("poll rest: " + JSON.stringify(rest));
+    if (!promise) throw "actionQueue empty when it shouldn't be.";
 
-    if (errno !== 0) {
-      if (self.onError)
-        self.onError(action.method, errno, node.fs.strerror(errno));
-      actionQueue = []; // empty the queue.
-      return;
+    var args = [];
+    for (var i = 0; i < arguments.length; i++) {
+      node.debug(JSON.stringify(arguments[i]));
+      args.push(arguments[i]);
     }
 
-    var rest = [];
-    for (var i = 1; i < arguments.length; i++)
-      rest.push(arguments[i]);
-
-    if (action.callback)
-      action.callback.apply(this, rest);
+    promise.emitSuccess(args);
 
     actionQueue.shift();
     act();
   }
 
-  function act () {
-    var action = actionQueue[0]; // peek at the head of the queue
-    if (action) {
-      internal_methods[action.method].apply(this, action.args);
+  function error () {
+    var promise = actionQueue[0];
+
+    if (!promise) throw "actionQueue empty when it shouldn't be.";
+
+    var args = [];
+    for (var i = 0; i < arguments.length; i++) {
+      args.push(arguments[i]);
     }
+
+    promise.emitError(args);
+    self.emitError(args);
   }
 
   var internal_methods = {
@@ -125,51 +138,63 @@ node.fs.File = function (options) {
           throw "Unknown mode";
       }
       // fix the mode here
-      node.fs.open(path, flags, 0666, function (status, fd) {
+      var promise = node.fs.open(path, flags, 0666);
+      
+      promise.addCallback(function (fd) {
         self.fd = fd;
-        poll(status, fd);
+        success(fd);
       });
+
+      promise.addErrback(error);
     },
 
     close: function ( ) {
-      node.fs.close(self.fd, function (status) {
+      var promise = node.fs.close(self.fd);
+
+      promise.addCallback(function () {
         self.fd = null;
-        poll(status);
+        success();
       });
+
+      promise.addErrback(error);
     }, 
 
     read: function (length, position) {
       //node.debug("encoding: " + self.encoding);
-      node.fs.read(self.fd, length, position, self.encoding, poll);
+      var promise = node.fs.read(self.fd, length, position, self.encoding);
+      promise.addCallback(success);  
+      promise.addErrback(error);  
     },
 
     write: function (data, position) {
-      node.fs.write(self.fd, data, position, poll);
+      var promise = node.fs.write(self.fd, data, position);
+      promise.addCallback(success);  
+      promise.addErrback(error);  
     }
   };
 
-  self.open = function (path, mode, callback) {
-    addAction("open", [path, mode], callback);
+  self.open = function (path, mode) {
+    return createAction("open", [path, mode]);
   };
    
-  self.close = function (callback) {
-    addAction("close", [], callback);
+  self.close = function () {
+    return createAction("close", []);
   };
 
-  self.read = function (length, pos, callback) {
-    addAction("read", [length, pos], callback);
+  self.read = function (length, pos) {
+    return createAction("read", [length, pos]);
   };
 
-  self.write = function (buf, pos, callback) {
-    addAction("write", [buf, pos], callback);
+  self.write = function (buf, pos) {
+    return createAction("write", [buf, pos]);
   };
 
-  self.print = function (data, callback) {
-    return self.write(data, null, callback);
+  self.print = function (data) {
+    return self.write(data, null);
   };
 
-  self.puts = function (data, callback) {
-    return self.write(data + "\n", null, callback);
+  self.puts = function (data) {
+    return self.write(data + "\n", null);
   };
 };
 
@@ -179,6 +204,4 @@ stdin  = new node.fs.File({ fd: node.STDIN_FILENO  });
 
 puts  = stdout.puts;
 print = stdout.print;
-p = function (data, callback) {
-  puts(JSON.stringify(data), callback);
-}
+p = function (data) { return puts(JSON.stringify(data)); }

src/http.js

@@ -537,20 +537,32 @@ function createClientRequest (connection, method, uri, header_lines) {
   return req;
 }
 
-node.http.cat = function (url, encoding, callback) {
+node.http.cat = function (url, encoding) {
+  var promise = new node.Promise();
+
   var uri = node.http.parseUri(url);
-  var req = node.http.createClient(uri.port || 80, uri.host).get(uri.path || "/");
+  var client = node.http.createClient(uri.port || 80, uri.host);
+  var req = client.get(uri.path || "/");
+
+  client.addListener("Error", function () {
+    promise.emitError();
+  });
+
+  var content = "";
+
   req.finish(function (res) {
-    var status = res.statusCode == 200 ? 0 : -1;
+    if (res.statusCode < 200 || res.statusCode >= 300) {
+      promise.emitError([res.statusCode]);
+      return;
+    }
     res.setBodyEncoding(encoding);
-    var content = "";
-    res.addListener("Body", function (chunk) {
-      content += chunk;
-    });
+    res.addListener("Body", function (chunk) { content += chunk; });
     res.addListener("BodyComplete", function () {
-      callback(status, content);
+      promise.emitSuccess([content]);
     });
   });
+
+  return promise;
 };
 
 })(); // anonymous namespace

src/node.h

@@ -44,7 +44,6 @@ public:
 
 protected:
   static void* Unwrap (v8::Handle<v8::Object> handle);
-  v8::Persistent<v8::Object> handle_;
 
   /* Attach() marks the object as being attached to an event loop.
    * Attached objects will not be garbage collected, even if
@@ -60,6 +59,7 @@ protected:
    * persistant handle.)
    */
   void Detach();
+  v8::Persistent<v8::Object> handle_;
 
 private:
   static void MakeWeak (v8::Persistent<v8::Value> _, void *data);

src/node.js

@@ -72,7 +72,7 @@ node.path = new function () {
 node.cat = function(location, encoding, callback) {
   var url_re = new RegExp("^http:\/\/");
   var f = url_re.exec(location) ? node.http.cat : node.fs.cat;
-  f(location, encoding, callback);
+  return f(location, encoding, callback);
 };
 
 // Module
@@ -105,12 +105,14 @@ node.Module.prototype.load = function (callback) {
     throw "Module '" + self.filename + "' is already loaded.";
   }
 
-  node.cat(self.filename, "utf8", function (status, content) {
-    if (status != 0) {
-      stderr.puts("Error reading " + self.filename);
-      node.exit(1);
-    }
+  var promise = node.cat(self.filename, "utf8");
+  
+  promise.addErrback(function () {
+    stderr.puts("Error reading " + self.filename);
+    node.exit(1);
+  });
 
+  promise.addCallback(function (content) {
     self.target.__require = function (path) { return self.newChild(path, {}); };
     self.target.__include = function (path) { self.newChild(path, self.target); };
 

test/mjsunit/test-file-cat-noexist.js

@@ -1,13 +1,23 @@
 include("mjsunit.js");
-var status = null;
+var got_error = false;
 
 function onLoad () {
   var dirname = node.path.dirname(__filename);
   var fixtures = node.path.join(dirname, "fixtures");
   var filename = node.path.join(fixtures, "does_not_exist.txt");
-  node.fs.cat(filename, "raw", function (s) { status = s });
+  var promise = node.fs.cat(filename, "raw");
+  
+  promise.addCallback(function (content) {
+    node.debug("cat returned some content: " + content);
+    node.debug("this shouldn't happen as the file doesn't exist...");
+    assertTrue(false);
+  });
+
+  promise.addErrback(function () {
+    got_error = true;
+  });
 }
 
 function onExit () {
-  assertTrue(status != 0);
+  assertTrue(got_error);
 }

test/mjsunit/test-file-open.js

@@ -10,14 +10,16 @@ function onLoad () {
   var x = node.path.join(fixtures, "x.txt");
 
   file = new node.fs.File;
-  file.onError = function () { got_error = true };
+  file.addListener("Error", function () { got_error = true });
 
-  file.open(x, "r", function () {
+  file.open(x, "r").addCallback(function () {
     opened = true
-    file.close(function () {
+    file.close().addCallback(function () {
       closed = true;
     });
   });
+
+  puts("hey!");
 }
 
 function onExit () {

test/mjsunit/test-http-cat.js

@@ -12,15 +12,26 @@ var server = node.http.createServer(function (req, res) {
 });
 server.listen(PORT);
 
+var got_good_server_content = false;
+var bad_server_got_error = false;
+
 function onLoad() {
-  node.http.cat("http://localhost:"+PORT, "utf8", function(status, content) {
-    assertEquals(body, content);
-    assertEquals(0, status)
-    server.close()
-  })
+  node.http.cat("http://localhost:"+PORT+"/", "utf8")
+    .addCallback(function (content) {
+      node.debug("got response");
+      got_good_server_content = true;
+      assertEquals(body, content);
+      server.close();
+    });
+
+  node.http.cat("http://localhost:12312/", "utf8")
+    .addErrback(function () {
+      node.debug("got error (this should happen)");
+      bad_server_got_error = true;
+    });
+}
 
-  node.http.cat("http://localhost:"+PORT+1, "utf8", function(status, content) {
-    assertEquals(-1, status)
-    assertEquals(nil, content)
-  })
+function onExit () {
+  assertTrue(got_good_server_content);
+  assertTrue(bad_server_got_error);
 }

test/mjsunit/test-http.js

@@ -36,6 +36,7 @@ function onLoad () {
     responses_recvd += 1;
     res.setBodyEncoding("utf8");
     res.addListener("Body", function (chunk) { body0 += chunk; });
+    node.debug("Got /hello response");
   });
 
   setTimeout(function () {
@@ -45,13 +46,18 @@ function onLoad () {
       responses_recvd += 1;
       res.setBodyEncoding("utf8");
       res.addListener("Body", function (chunk) { body1 += chunk; });
+      node.debug("Got /world response");
     });
   }, 1);
 }
 
 function onExit () {
+  node.debug("responses_recvd: " + responses_recvd);
   assertEquals(2, responses_recvd);
+
+  node.debug("responses_sent: " + responses_sent);
   assertEquals(2, responses_sent);
+
   assertEquals("The path was /hello", body0);
   assertEquals("The path was /world", body1);
 }

test/mjsunit/test-node-cat.js

@@ -12,18 +12,39 @@ var server = node.http.createServer(function (req, res) {
 });
 server.listen(PORT);
 
+var errors = 0;
+var successes = 0;
+
 function onLoad() {
-  node.cat("http://localhost:"+PORT, "utf8", function(status, content) {
+  var promise = node.cat("http://localhost:"+PORT, "utf8");
+  
+  promise.addCallback(function (content) {
     assertEquals(body, content);
-    assertEquals(0, status)
     server.close()
-  })
+    successes += 1;
+  });
+
+  promise.addErrback(function () {
+    errors += 1;
+  });
   
   var dirname = node.path.dirname(__filename);
   var fixtures = node.path.join(dirname, "fixtures");
   var x = node.path.join(fixtures, "x.txt");
-  node.cat(x, "utf8", function(status, content) {
-    assertEquals(0, status)
+
+  promise = node.cat(x, "utf8");
+  
+  promise.addCallback(function (content) {
     assertEquals("xyz", content.replace(/[\r\n]/, ''))
-  })
+    successes += 1;
+  });
+
+  promise.addErrback(function () {
+    errors += 1;
+  });
+}
+
+function onExit () {
+  assertEquals(2, successes);
+  assertEquals(0, errors);
 }