fix uv_async usage

lib/sqlite3.js

@@ -3,9 +3,6 @@ var path = require('path');
 var util = require('util');
 var EventEmitter = require('events').EventEmitter;
 
-var Database = sqlite3.Database;
-var Statement = sqlite3.Statement;
-
 function errorCallback(args) {
     if (typeof args[args.length - 1] === 'function') {
         var callback = args[args.length - 1];
@@ -13,6 +10,11 @@ function errorCallback(args) {
     }
 }
 
+function inherits(target, source) {
+  for (var k in source.prototype)
+    target.prototype[k] = source.prototype[k];
+}
+
 sqlite3.cached = {
     Database: function(file, a, b) {
         if (file === '' || file === ':memory:') {
@@ -43,6 +45,13 @@ sqlite3.cached = {
     objects: {}
 };
 
+
+var Database = sqlite3.Database;
+var Statement = sqlite3.Statement;
+
+inherits(Database, EventEmitter);
+inherits(Statement, EventEmitter);
+
 // Database#prepare(sql, [bind1, bind2, ...], [callback])
 Database.prototype.prepare = function(sql) {
     var params = Array.prototype.slice.call(arguments, 1);
@@ -149,12 +158,6 @@ Database.prototype.removeAllListeners = function(type) {
     return val;
 };
 
-Database.prototype.emit = EventEmitter.prototype.emit;
-Database.prototype.once = EventEmitter.prototype.once;
-
-// util.inherits(Database, EventEmitter);
-// util.inherits(Statement, EventEmitter);
-
 // Save the stack trace over EIO callbacks.
 sqlite3.verbose = function() {
     if (!isVerbose) {

src/statement.cc

@@ -607,7 +607,6 @@ void Statement::EIO_Each(eio_req *req) {
     STATEMENT_INIT(EachBaton);
 
     Async* async = baton->async;
-    fprintf(stderr, "async:%p\n", async);
 
     sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->handle);
 
@@ -620,7 +619,6 @@ void Statement::EIO_Each(eio_req *req) {
 
     if (stmt->Bind(baton->parameters)) {
         while (true) {
-            fprintf(stderr, "before mutex\n");
             sqlite3_mutex_enter(mtx);
             stmt->status = sqlite3_step(stmt->handle);
             if (stmt->status == SQLITE_ROW) {
@@ -628,33 +626,32 @@ void Statement::EIO_Each(eio_req *req) {
                 Row* row = new Row();
                 GetRow(row, stmt->handle);
 
-                // pthread_mutex_lock(&async->mutex);
+                pthread_mutex_lock(&async->mutex);
                 async->data.push_back(row);
                 retrieved++;
-                // pthread_mutex_unlock(&async->mutex);
-                
-                fprintf(stderr, "retrieved:%d\n", retrieved);
-                // uv_async_send(&async->watcher);
+                pthread_mutex_unlock(&async->mutex);
+
+                uv_async_send(&async->watcher);
             }
             else {
                 if (stmt->status != SQLITE_DONE) {
                     stmt->message = std::string(sqlite3_errmsg(stmt->db->handle));
                 }
                 sqlite3_mutex_leave(mtx);
-                fprintf(stderr, "done\n");
                 break;
             }
         }
     }
-    fprintf(stderr, "retrieved:%d\n", retrieved);
 
     async->completed = true;
-    // uv_async_send(&async->watcher);
+    uv_async_send(&async->watcher);
 }
 
 void Statement::CloseCallback(uv_handle_t* handle) {
     assert(handle != NULL);
-    fprintf(stderr, "close callback\n");
+    Async* async = static_cast<Async*>(handle->data);
+    delete async;
+    handle->data = NULL;
 }
 
 void Statement::AsyncEach(uv_async_t* handle, int status) {
@@ -680,16 +677,15 @@ void Statement::AsyncEach(uv_async_t* handle, int status) {
             Rows::const_iterator it = rows.begin();
             Rows::const_iterator end = rows.end();
             for (int i = 0; it < end; it++, i++) {
-                // argv[1] = RowToJS(*it);
+                argv[1] = RowToJS(*it);
                 async->retrieved++;
-            //     TRY_CATCH_CALL(async->stmt->handle_, baton->callback, 2, argv);
-                // delete *it;
+                TRY_CATCH_CALL(async->stmt->handle_, baton->callback, 2, argv);
+                delete *it;
             }
         }
     }
 
     if (async->completed) {
-        fprintf(stderr, "completed\n");
         if (!baton->completed.IsEmpty() &&
                 baton->completed->IsFunction()) {
             Local<Value> argv[] = {
@@ -698,9 +694,7 @@ void Statement::AsyncEach(uv_async_t* handle, int status) {
             };
             TRY_CATCH_CALL(async->stmt->handle_, baton->completed, 2, argv);
         }
-        // uv_close((uv_handle_t*)handle, CloseCallback);
-        delete async;
-        handle->data = NULL;
+        uv_close((uv_handle_t*)handle, CloseCallback);
     }
 }
 

src/statement.h

@@ -165,11 +165,8 @@ public:
                 stmt(st), baton(eb), completed(false), retrieved(0) {
             watcher.data = this;
             pthread_mutex_init(&mutex, NULL);
-            fprintf(stderr, "initialized mutex\n");
             stmt->Ref();
-            fprintf(stderr, "referenced stmt\n");
             uv_async_init(uv_default_loop(), &watcher, async_cb);
-            fprintf(stderr, "started async\n");
         }
 
         ~Async() {

test/each.test.js

@@ -19,24 +19,24 @@ exports['test Statement#each'] = function(beforeExit) {
         assert.equal(retrieved, total, "Only retrieved " + retrieved + " out of " + total + " rows.");
     });
 };
-// 
-// exports['test Statement#each with complete callback'] = function(beforeExit) {
-//     var db = new sqlite3.Database('test/support/big.db', sqlite3.OPEN_READONLY);
-// 
-//     var total = 10000;
-//     var retrieved = 0;
-//     var completed = false;
-// 
-//     db.each('SELECT id, txt FROM foo LIMIT 0, ?', total, function(err, row) {
-//         if (err) throw err;
-//         retrieved++;
-//     }, function(err, num) {
-//         assert.equal(retrieved, num);
-//         completed = true;
-//     });
-// 
-//     beforeExit(function() {
-//         assert.ok(completed);
-//         assert.equal(retrieved, total, "Only retrieved " + retrieved + " out of " + total + " rows.");
-//     });
-// };
+
+exports['test Statement#each with complete callback'] = function(beforeExit) {
+    var db = new sqlite3.Database('test/support/big.db', sqlite3.OPEN_READONLY);
+
+    var total = 10000;
+    var retrieved = 0;
+    var completed = false;
+
+    db.each('SELECT id, txt FROM foo LIMIT 0, ?', total, function(err, row) {
+        if (err) throw err;
+        retrieved++;
+    }, function(err, num) {
+        assert.equal(retrieved, num);
+        completed = true;
+    });
+
+    beforeExit(function() {
+        assert.ok(completed);
+        assert.equal(retrieved, total, "Only retrieved " + retrieved + " out of " + total + " rows.");
+    });
+};