Merge pull request #130 from Sage/reader

Reader

Author: joeferner

lib/oracle.js

@@ -35,21 +35,36 @@ exports.OCCITIMESTAMP = 7;
 exports.OCCINUMBER = 8;
 exports.OCCIBLOB = 9;
 
-// Reader.prototype.nextRow is implemented in JS rather than C++.
-// This is easier and also more efficient because we don't cross the JS/C++ boundary every time
-// we read a record.
-bindings.Reader.prototype.nextRow = function(cb) {
+// Reader is implemented in JS around a C++ handle
+// This is easier and also more efficient because we don't cross the JS/C++ boundary 
+// every time we read a record.
+function Reader(handle) {
+  this._handle = handle;
+  this._error = null;
+  this._rows = [];
+}
+
+Reader.prototype.nextRows = function() {
+  this._handle.nextRows.apply(this._handle, arguments);
+}
+
+Reader.prototype.nextRow = function(cb) {
   var self = this;
-  if (self._error || (self._rows && self._rows.length > 0)) {
+  if (!self._handle || self._error || (self._rows && self._rows.length > 0)) {
     process.nextTick(function() {
       cb(self._error, self._rows && self._rows.shift());
     });
   } else {
     // nextRows willl use the prefetch row count as window size
-    self.nextRows(function(err, result) {
+    self._handle.nextRows(function(err, result) {
       self._error = err || self._error;
       self._rows = result;
+      if (err || result.length === 0) self._handle = null;
       cb(self._error, self._rows && self._rows.shift());
     });
   }
 };
+
+bindings.Connection.prototype.reader = function(sql, args) {
+  return new Reader(this.readerHandle(sql, args));
+}

src/connection.cpp

@@ -22,7 +22,7 @@ void Connection::Init(Handle<Object> target) {
 
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "execute", Execute);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "executeSync", ExecuteSync);
-  NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "reader", CreateReader);
+  NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "readerHandle", CreateReader);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "close", Close);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "isConnected", IsConnected);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "setAutoCommit", SetAutoCommit);
@@ -391,7 +391,7 @@ void Connection::EIO_AfterCommit(uv_work_t* req, int status) {
   argv[0] = Undefined();
   node::MakeCallback(Context::GetCurrent()->Global(), uni::Deref(baton->callback), 2, argv);
   delete baton;
-
+  delete req;
 }
 
 void Connection::EIO_Rollback(uv_work_t* req) {
@@ -410,7 +410,7 @@ void Connection::EIO_AfterRollback(uv_work_t* req, int status) {
   argv[0] = Undefined();
   node::MakeCallback(Context::GetCurrent()->Global(), uni::Deref(baton->callback), 2, argv);
   delete baton;
-
+  delete req;
 }
 
 void Connection::EIO_Execute(uv_work_t* req) {
@@ -705,6 +705,7 @@ void Connection::EIO_AfterExecute(uv_work_t* req, int status) {
   }
 
   delete baton;
+  delete req;
 }
 
 void Connection::handleResult(ExecuteBaton* baton, Handle<Value> (&argv)[2]) {

src/connection.h

@@ -36,6 +36,10 @@ namespace uni {
   Handle<T> Deref(Persistent<T>& handle) {
     return Handle<T>::New(Isolate::GetCurrent(), handle);
   }
+  template <class T>
+  Local<T> HandleToLocal(Handle<T> handle) {
+    return handle;
+  }
   inline Handle<Value> BufferToHandle(BufferType buf) {
     return buf;
   }
@@ -57,6 +61,10 @@ namespace uni {
   Handle<T> Deref(Persistent<T>& handle) {
     return Local<T>::New(handle);
   }
+  template <class T>
+  Local<T> HandleToLocal(Handle<T> handle) {
+    return Local<T>::New(handle);
+  }
   inline Handle<Value> BufferToHandle(BufferType buf) {
     return buf->handle_;
   }

src/executeBaton.cpp

@@ -33,14 +33,7 @@ ExecuteBaton::~ExecuteBaton() {
     delete val;
   }
 
-  if(rows) {
-    for (std::vector<row_t*>::iterator iterator = rows->begin(), end = rows->end(); iterator != end; ++iterator) {
-      row_t* currentRow = *iterator;
-      delete currentRow;
-    }
-
-    delete rows;
-  }
+  ResetRows();
 
   if(outputs) {
     for (std::vector<output_t*>::iterator iterator = outputs->begin(), end = outputs->end(); iterator != end; ++iterator) {
@@ -53,6 +46,18 @@ ExecuteBaton::~ExecuteBaton() {
   if(error) delete error;
 }
 
+void ExecuteBaton::ResetRows() {
+  if(rows) {
+    for (std::vector<row_t*>::iterator iterator = rows->begin(), end = rows->end(); iterator != end; ++iterator) {
+      row_t* currentRow = *iterator;
+      delete currentRow;
+    }
+
+    delete rows;
+    rows = NULL;
+  }
+}
+
 double CallDateMethod(v8::Local<v8::Date> date, const char* methodName) {
   Handle<Value> args[1]; // should be zero but on windows the compiler will not allow a zero length array
   Local<Value> result = Local<Function>::Cast(date->Get(String::New(methodName)))->Call(date, 0, args);

src/executeBaton.h

@@ -61,6 +61,7 @@ class ExecuteBaton {
 public:
   ExecuteBaton(Connection* connection, const char* sql, v8::Local<v8::Array>* values, v8::Handle<v8::Function>* callback);
   ~ExecuteBaton();
+  void ResetRows();
 
   Connection *connection;
   v8::Persistent<v8::Function> callback;

src/oracle_bindings.cpp

@@ -148,6 +148,7 @@ void OracleClient::EIO_AfterConnect(uv_work_t* req, int status) {
   node::MakeCallback(Context::GetCurrent()->Global(), uni::Deref(baton->callback), 2, argv);
 
   delete baton;
+  delete req;
 }
 
 uni::CallbackType OracleClient::ConnectSync(const uni::FunctionCallbackInfo& args) {

src/reader.cpp

@@ -48,16 +48,20 @@ uni::CallbackType Reader::NextRows(const uni::FunctionCallbackInfo& args) {
     Local<String> message = String::New(baton->error->c_str());
     UNI_THROW(Exception::Error(message));
   }
+  if (baton->busy) {
+    UNI_THROW(Exception::Error(String::New("invalid state: reader is busy with another nextRows call")));
+  }
+  baton->busy = true;
 
   if (args.Length() > 1) {
     REQ_INT_ARG(0, count);
     REQ_FUN_ARG(1, callback);
     baton->count = count;
-    uni::Reset(baton->nextRowsCallback, callback);
+    uni::Reset(baton->callback, callback);
   } else {
     REQ_FUN_ARG(0, callback);
     baton->count = baton->connection->getPrefetchRowCount();
-    uni::Reset(baton->nextRowsCallback, callback);
+    uni::Reset(baton->callback, callback);
   }
   if (baton->count <= 0) baton->count = 1;
 
@@ -72,77 +76,64 @@ uni::CallbackType Reader::NextRows(const uni::FunctionCallbackInfo& args) {
 void Reader::EIO_NextRows(uv_work_t* req) {
   ReaderBaton* baton = static_cast<ReaderBaton*>(req->data);
 
-  baton->rows = NULL;
-  baton->error = NULL;
+  baton->rows = new vector<row_t*>();
+  if (baton->done) return;
 
-  try {
-    if(! baton->connection->getConnection()) {
-      baton->error = new std::string("Connection already closed");
-      return;
-    }
-    if (!baton->rs) {
+  if (!baton->connection->getConnection()) {
+    baton->error = new std::string("Connection already closed");
+    return;
+  }
+  if (!baton->rs) {
+    try {
       baton->stmt = baton->connection->getConnection()->createStatement(baton->sql);
       baton->stmt->setAutoCommit(baton->connection->getAutoCommit());
       baton->stmt->setPrefetchRowCount(baton->count);
       Connection::SetValuesOnStatement(baton->stmt, baton);
-      if (baton->error) goto cleanup;
+      if (baton->error) return;
 
       int status = baton->stmt->execute();
-      if(status != oracle::occi::Statement::RESULT_SET_AVAILABLE) {
+      if (status != oracle::occi::Statement::RESULT_SET_AVAILABLE) {
          baton->error = new std::string("Connection already closed");
         return;
       }     
       baton->rs = baton->stmt->getResultSet();
-      Connection::CreateColumnsFromResultSet(baton->rs, baton, baton->columns);
-      if (baton->error) goto cleanup;
-    }
-    baton->rows = new vector<row_t*>();
-
-    for (int i = 0; i < baton->count && baton->rs->next(); i++) {
-      row_t* row = Connection::CreateRowFromCurrentResultSetRow(baton->rs, baton, baton->columns);
-      if (baton->error) goto cleanup;
-      baton->rows->push_back(row);
+    } catch (oracle::occi::SQLException &ex) {
+      baton->error = new string(ex.getMessage());
+      return;
     }
-  } catch(oracle::occi::SQLException &ex) {
-    baton->error = new string(ex.getMessage());
-  } catch (const exception& ex) {
-    baton->error = new string(ex.what());
-  } catch (...) {
-    baton->error = new string("Unknown Error");
+    Connection::CreateColumnsFromResultSet(baton->rs, baton, baton->columns);
+    if (baton->error) return;
   }
-cleanup:
-  // nothing for now, cleanup happens in destructor
-  ;
-}
-
-#if NODE_MODULE_VERSION >= 0x000D
-void ReaderWeakReferenceCallback(Isolate* isolate, v8::Persistent<v8::Function>* callback, void* dummy)
-{
-  callback->Dispose();
-}
-#else
-void ReaderWeakReferenceCallback(v8::Persistent<v8::Value> callback, void* dummy)
-{
-  (Persistent<Function>(Function::Cast(*callback))).Dispose();
+  for (int i = 0; i < baton->count && baton->rs->next(); i++) {
+    row_t* row = Connection::CreateRowFromCurrentResultSetRow(baton->rs, baton, baton->columns);
+    if (baton->error) return;
+    baton->rows->push_back(row);
+  }
+  if (baton->rows->size() < (size_t)baton->count) baton->done = true;
 }
-#endif
 
 void Reader::EIO_AfterNextRows(uv_work_t* req, int status) {
   UNI_SCOPE(scope);
   ReaderBaton* baton = static_cast<ReaderBaton*>(req->data);
 
+  baton->busy = false;
   baton->connection->Unref();
-
-  try {
-    Handle<Value> argv[2];
-    Connection::handleResult(baton, argv);
-    node::MakeCallback(Context::GetCurrent()->Global(), uni::Deref(baton->nextRowsCallback), 2, argv);
-  } catch(const exception &ex) {
-    Handle<Value> argv[2];
-    argv[0] = Exception::Error(String::New(ex.what()));
-    argv[1] = Undefined();
-    node::MakeCallback(Context::GetCurrent()->Global(), uni::Deref(baton->nextRowsCallback), 2, argv);
+  // transfer callback to local and dispose persistent handle
+  // must be done before invoking callback because callback may set another callback into baton->callback
+  Local<Function> cb = uni::HandleToLocal(uni::Deref(baton->callback));
+  baton->callback.Dispose();
+  baton->callback.Clear();
+
+  Handle<Value> argv[2];
+  Connection::handleResult(baton, argv);
+  node::MakeCallback(Context::GetCurrent()->Global(), cb, 2, argv);
+  
+  baton->ResetRows();
+  if (baton->done || baton->error) {
+    // free occi resources so that we don't run out of cursors if gc is not fast enough
+    // reader destructor will delete the baton and everything else.
+    baton->ResetStatement();
   }
-  baton->nextRowsCallback.MakeWeak((void*)NULL, ReaderWeakReferenceCallback);
+  delete req;
 }
 

src/readerBaton.h

@@ -10,10 +10,16 @@ class ReaderBaton : public ExecuteBaton {
   ReaderBaton(Connection* connection, const char* sql, v8::Local<v8::Array>* values) : ExecuteBaton(connection, sql, values, NULL) {
     stmt = NULL;
     rs = NULL;
+    done = false;
+    busy = false;
   }
   ~ReaderBaton() {
-	if(stmt && rs) {
-	  stmt->closeResultSet(rs);
+    ResetStatement();
+  }
+
+  void ResetStatement() {
+	  if(stmt && rs) {
+      stmt->closeResultSet(rs);
       rs = NULL;
     }
     if(stmt) {
@@ -24,10 +30,11 @@ class ReaderBaton : public ExecuteBaton {
     }
   }
 
-  v8::Persistent<v8::Function> nextRowsCallback;  
   oracle::occi::Statement* stmt;
   oracle::occi::ResultSet* rs;
   int count;
+  bool done;
+  bool busy;
 };
 
 #endif