NanScope();

Local<FunctionTemplate> t = FunctionTemplate::New(New);

t->InstanceTemplate()->SetInternalFieldCount(1);

t->SetClassName(String::NewSymbol("Statement"));

NODE_SET_PROTOTYPE_METHOD(t, "bind", Bind);

NODE_SET_PROTOTYPE_METHOD(t, "get", Get);

NODE_SET_PROTOTYPE_METHOD(t, "run", Run);

NODE_SET_PROTOTYPE_METHOD(t, "all", All);

NODE_SET_PROTOTYPE_METHOD(t, "each", Each);

NODE_SET_PROTOTYPE_METHOD(t, "reset", Reset);

NODE_SET_PROTOTYPE_METHOD(t, "finalize", Finalize);

NanAssignPersistent(FunctionTemplate, constructor_template, t);

target->Set(String::NewSymbol("Statement"),

t->GetFunction());

if (finalized && !queue.empty()) {

return CleanQueue();

}

while (prepared && !locked && !queue.empty()) {

Call* call = queue.front();

queue.pop();

call->callback(call->baton);

delete call;

}

if (finalized) {

queue.push(new Call(callback, baton));

CleanQueue();

}

else if (!prepared || locked) {

queue.push(new Call(callback, baton));

}

else {

callback(baton);

}

Statement* stmt = baton->stmt;

// Fail hard on logic errors.

assert(stmt->status != 0);

EXCEPTION(String::New(stmt->message.c_str()), stmt->status, exception);

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

Local<Value> argv[] = { exception };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 1, argv);

}

else {

Local<Value> argv[] = { String::NewSymbol("error"), exception };

EMIT_EVENT(NanObjectWrapHandle(stmt), 2, argv);

}

NanScope();

if (!args.IsConstructCall()) {

return NanThrowTypeError("Use the new operator to create new Statement objects");

}

int length = args.Length();

if (length <= 0 || !Database::HasInstance(args[0])) {

return NanThrowTypeError("Database object expected");

}

else if (length <= 1 || !args[1]->IsString()) {

return NanThrowTypeError("SQL query expected");

}

else if (length > 2 && !args[2]->IsUndefined() && !args[2]->IsFunction()) {

return NanThrowTypeError("Callback expected");

}

Database* db = ObjectWrap::Unwrap<Database>(args[0]->ToObject());

Local<String> sql = Local<String>::Cast(args[1]);

args.This()->Set(String::NewSymbol("sql"), sql, ReadOnly);

Statement* stmt = new Statement(db);

stmt->Wrap(args.This());

PrepareBaton* baton = new PrepareBaton(db, Local<Function>::Cast(args[2]), stmt);

baton->sql = std::string(*String::Utf8Value(sql));

db->Schedule(Work_BeginPrepare, baton);

NanReturnValue(args.This());

assert(baton->db->open);

baton->db->pending++;

int status = uv_queue_work(uv_default_loop(),

&baton->request, Work_Prepare, (uv_after_work_cb)Work_AfterPrepare);

assert(status == 0);

STATEMENT_INIT(PrepareBaton);

// In case preparing fails, we use a mutex to make sure we get the associated

// error message.

sqlite3_mutex* mtx = sqlite3_db_mutex(baton->db->_handle);

sqlite3_mutex_enter(mtx);

stmt->status = sqlite3_prepare_v2(

baton->db->_handle,

baton->sql.c_str(),

baton->sql.size(),

&stmt->_handle,

NULL

);

if (stmt->status != SQLITE_OK) {

stmt->message = std::string(sqlite3_errmsg(baton->db->_handle));

stmt->_handle = NULL;

}

sqlite3_mutex_leave(mtx);

NanScope();

STATEMENT_INIT(PrepareBaton);

if (stmt->status != SQLITE_OK) {

Error(baton);

stmt->Finalize();

}

else {

stmt->prepared = true;

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

Local<Value> argv[] = { NanNewLocal<Value>(Null()) };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 1, argv);

}

}

STATEMENT_END();

if (parameters.size() == 0) {

return true;

}

sqlite3_reset(_handle);

sqlite3_clear_bindings(_handle);

Parameters::const_iterator it = parameters.begin();

Parameters::const_iterator end = parameters.end();

for (; it < end; ++it) {

Values::Field* field = *it;

if (field != NULL) {

int pos;

if (field->index > 0) {

pos = field->index;

}

else {

pos = sqlite3_bind_parameter_index(_handle, field->name.c_str());

}

switch (field->type) {

case SQLITE_INTEGER: {

status = sqlite3_bind_int(_handle, pos,

((Values::Integer*)field)->value);

} break;

case SQLITE_FLOAT: {

status = sqlite3_bind_double(_handle, pos,

((Values::Float*)field)->value);

} break;

case SQLITE_TEXT: {

status = sqlite3_bind_text(_handle, pos,

((Values::Text*)field)->value.c_str(),

((Values::Text*)field)->value.size(), SQLITE_TRANSIENT);

} break;

case SQLITE_BLOB: {

status = sqlite3_bind_blob(_handle, pos,

((Values::Blob*)field)->value,

((Values::Blob*)field)->length, SQLITE_TRANSIENT);

} break;

case SQLITE_NULL: {

status = sqlite3_bind_null(_handle, pos);

} break;

}

}

if (status != SQLITE_OK) {

message = std::string(sqlite3_errmsg(db->_handle));

return false;

}

}

return true;

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

Baton* baton = stmt->Bind<Baton>(args);

if (baton == NULL) {

return NanThrowTypeError("Data type is not supported");

}

else {

stmt->Schedule(Work_BeginBind, baton);

NanReturnValue(args.This());

}

STATEMENT_INIT(Baton);

sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->_handle);

sqlite3_mutex_enter(mtx);

stmt->Bind(baton->parameters);

sqlite3_mutex_leave(mtx);

NanScope();

STATEMENT_INIT(Baton);

if (stmt->status != SQLITE_OK) {

Error(baton);

}

else {

// Fire callbacks.

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

Local<Value> argv[] = { NanNewLocal<Value>(Null()) };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 1, argv);

}

}

STATEMENT_END();

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

Baton* baton = stmt->Bind<RowBaton>(args);

if (baton == NULL) {

return NanThrowError("Data type is not supported");

}

else {

stmt->Schedule(Work_BeginGet, baton);

NanReturnValue(args.This());

}

STATEMENT_INIT(RowBaton);

if (stmt->status != SQLITE_DONE || baton->parameters.size()) {

sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->_handle);

sqlite3_mutex_enter(mtx);

if (stmt->Bind(baton->parameters)) {

stmt->status = sqlite3_step(stmt->_handle);

if (!(stmt->status == SQLITE_ROW || stmt->status == SQLITE_DONE)) {

stmt->message = std::string(sqlite3_errmsg(stmt->db->_handle));

}

}

sqlite3_mutex_leave(mtx);

if (stmt->status == SQLITE_ROW) {

// Acquire one result row before returning.

GetRow(&baton->row, stmt->_handle);

}

}

NanScope();

STATEMENT_INIT(RowBaton);

if (stmt->status != SQLITE_ROW && stmt->status != SQLITE_DONE) {

Error(baton);

}

else {

// Fire callbacks.

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

if (stmt->status == SQLITE_ROW) {

// Create the result array from the data we acquired.

Local<Value> argv[] = { NanNewLocal<Value>(Null()), RowToJS(&baton->row) };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 2, argv);

}

else {

Local<Value> argv[] = { NanNewLocal<Value>(Null()) };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 1, argv);

}

}

}

STATEMENT_END();

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

Baton* baton = stmt->Bind<RunBaton>(args);

if (baton == NULL) {

return NanThrowError("Data type is not supported");

}

else {

stmt->Schedule(Work_BeginRun, baton);

NanReturnValue(args.This());

}

STATEMENT_INIT(RunBaton);

sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->_handle);

sqlite3_mutex_enter(mtx);

// Make sure that we also reset when there are no parameters.

if (!baton->parameters.size()) {

sqlite3_reset(stmt->_handle);

}

if (stmt->Bind(baton->parameters)) {

stmt->status = sqlite3_step(stmt->_handle);

if (!(stmt->status == SQLITE_ROW || stmt->status == SQLITE_DONE)) {

stmt->message = std::string(sqlite3_errmsg(stmt->db->_handle));

}

else {

baton->inserted_id = sqlite3_last_insert_rowid(stmt->db->_handle);

baton->changes = sqlite3_changes(stmt->db->_handle);

}

}

sqlite3_mutex_leave(mtx);

NanScope();

STATEMENT_INIT(RunBaton);

if (stmt->status != SQLITE_ROW && stmt->status != SQLITE_DONE) {

Error(baton);

}

else {

// Fire callbacks.

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

NanObjectWrapHandle(stmt)->Set(String::NewSymbol("lastID"), Local<Integer>(Integer::New(baton->inserted_id)));

NanObjectWrapHandle(stmt)->Set(String::NewSymbol("changes"), Local<Integer>(Integer::New(baton->changes)));

Local<Value> argv[] = { NanNewLocal<Value>(Null()) };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 1, argv);

}

}

STATEMENT_END();

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

Baton* baton = stmt->Bind<RowsBaton>(args);

if (baton == NULL) {

return NanThrowError("Data type is not supported");

}

else {

stmt->Schedule(Work_BeginAll, baton);

NanReturnValue(args.This());

}

STATEMENT_INIT(RowsBaton);

sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->_handle);

sqlite3_mutex_enter(mtx);

// Make sure that we also reset when there are no parameters.

if (!baton->parameters.size()) {

sqlite3_reset(stmt->_handle);

}

if (stmt->Bind(baton->parameters)) {

while ((stmt->status = sqlite3_step(stmt->_handle)) == SQLITE_ROW) {

Row* row = new Row();

GetRow(row, stmt->_handle);

baton->rows.push_back(row);

}

if (stmt->status != SQLITE_DONE) {

stmt->message = std::string(sqlite3_errmsg(stmt->db->_handle));

}

}

sqlite3_mutex_leave(mtx);

NanScope();

STATEMENT_INIT(RowsBaton);

if (stmt->status != SQLITE_DONE) {

Error(baton);

}

else {

// Fire callbacks.

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

if (baton->rows.size()) {

// Create the result array from the data we acquired.

Local<Array> result(Array::New(baton->rows.size()));

Rows::const_iterator it = baton->rows.begin();

Rows::const_iterator end = baton->rows.end();

for (int i = 0; it < end; ++it, i++) {

result->Set(i, RowToJS(*it));

delete *it;

}

Local<Value> argv[] = { NanNewLocal<Value>(Null()), result };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 2, argv);

}

else {

// There were no result rows.

Local<Value> argv[] = {

NanNewLocal<Value>(Null()),

NanNewLocal<Value>(Array::New(0))

};

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 2, argv);

}

}

}

STATEMENT_END();

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

int last = args.Length();

Local<Function> completed;

if (last >= 2 && args[last - 1]->IsFunction() && args[last - 2]->IsFunction()) {

completed = Local<Function>::Cast(args[--last]);

}

EachBaton* baton = stmt->Bind<EachBaton>(args, 0, last);

if (baton == NULL) {

return NanThrowError("Data type is not supported");

}

else {

NanAssignPersistent(Function, baton->completed, completed);

stmt->Schedule(Work_BeginEach, baton);

NanReturnValue(args.This());

}

// Only create the Async object when we're actually going into

// the event loop. This prevents dangling events.

EachBaton* each_baton = static_cast<EachBaton*>(baton);

each_baton->async = new Async(each_baton->stmt, AsyncEach);

NanAssignPersistent(Function, each_baton->async->item_cb, each_baton->callback);

NanAssignPersistent(Function, each_baton->async->completed_cb, each_baton->completed);

STATEMENT_BEGIN(Each);

STATEMENT_INIT(EachBaton);

Async* async = baton->async;

sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->_handle);

int retrieved = 0;

// Make sure that we also reset when there are no parameters.

if (!baton->parameters.size()) {

sqlite3_reset(stmt->_handle);

}

if (stmt->Bind(baton->parameters)) {

while (true) {

sqlite3_mutex_enter(mtx);

stmt->status = sqlite3_step(stmt->_handle);

if (stmt->status == SQLITE_ROW) {

sqlite3_mutex_leave(mtx);

Row* row = new Row();

GetRow(row, stmt->_handle);

NODE_SQLITE3_MUTEX_LOCK(&async->mutex)

async->data.push_back(row);

retrieved++;

NODE_SQLITE3_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);

break;

}

}

}

async->completed = true;

uv_async_send(&async->watcher);

assert(handle != NULL);

assert(handle->data != NULL);

Async* async = static_cast<Async*>(handle->data);

delete async;

NanScope();

Async* async = static_cast<Async*>(handle->data);

while (true) {

// Get the contents out of the data cache for us to process in the JS callback.

Rows rows;

NODE_SQLITE3_MUTEX_LOCK(&async->mutex)

rows.swap(async->data);

NODE_SQLITE3_MUTEX_UNLOCK(&async->mutex)

if (rows.empty()) {

break;

}

Local<Function> cb = NanPersistentToLocal(async->item_cb);

if (!cb.IsEmpty() && cb->IsFunction()) {

Local<Value> argv[2];

argv[0] = NanNewLocal<Value>(Null());

Rows::const_iterator it = rows.begin();

Rows::const_iterator end = rows.end();

for (int i = 0; it < end; ++it, i++) {

argv[1] = RowToJS(*it);

async->retrieved++;

TRY_CATCH_CALL(NanObjectWrapHandle(async->stmt), cb, 2, argv);

delete *it;

}

}

}

Local<Function> cb = NanPersistentToLocal(async->completed_cb);

if (async->completed) {

if (!cb.IsEmpty() &&

cb->IsFunction()) {

Local<Value> argv[] = {

NanNewLocal<Value>(Null()),

Integer::New(async->retrieved)

};

TRY_CATCH_CALL(NanObjectWrapHandle(async->stmt), cb, 2, argv);

}

uv_close((uv_handle_t*)handle, CloseCallback);

}

NanScope();

STATEMENT_INIT(EachBaton);

if (stmt->status != SQLITE_DONE) {

Error(baton);

}

STATEMENT_END();

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

OPTIONAL_ARGUMENT_FUNCTION(0, callback);

Baton* baton = new Baton(stmt, callback);

stmt->Schedule(Work_BeginReset, baton);

NanReturnValue(args.This());

STATEMENT_INIT(Baton);

sqlite3_reset(stmt->_handle);

stmt->status = SQLITE_OK;

NanScope();

STATEMENT_INIT(Baton);

// Fire callbacks.

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

Local<Value> argv[] = { NanNewLocal<Value>(Null()) };

TRY_CATCH_CALL(NanObjectWrapHandle(stmt), cb, 1, argv);

}

STATEMENT_END();

Local<Object> result(Object::New());

Row::const_iterator it = row->begin();

Row::const_iterator end = row->end();

for (int i = 0; it < end; ++it, i++) {

Values::Field* field = *it;

Local<Value> value;

switch (field->type) {

case SQLITE_INTEGER: {

value = Local<Value>(Number::New(((Values::Integer*)field)->value));

} break;

case SQLITE_FLOAT: {

value = Local<Value>(Number::New(((Values::Float*)field)->value));

} break;

case SQLITE_TEXT: {

value = Local<Value>(String::New(((Values::Text*)field)->value.c_str(), ((Values::Text*)field)->value.size()));

} break;

case SQLITE_BLOB: {

value = NanNewLocal<Value>(NanNewBufferHandle(((Values::Blob*)field)->value, ((Values::Blob*)field)->length));

} break;

case SQLITE_NULL: {

value = NanNewLocal<Value>(Null());

} break;

}

result->Set(String::NewSymbol(field->name.c_str()), value);

DELETE_FIELD(field);

}

return result;

int rows = sqlite3_column_count(stmt);

for (int i = 0; i < rows; i++) {

int type = sqlite3_column_type(stmt, i);

const char* name = sqlite3_column_name(stmt, i);

switch (type) {

case SQLITE_INTEGER: {

row->push_back(new Values::Integer(name, sqlite3_column_int64(stmt, i)));

} break;

case SQLITE_FLOAT: {

row->push_back(new Values::Float(name, sqlite3_column_double(stmt, i)));

} break;

case SQLITE_TEXT: {

const char* text = (const char*)sqlite3_column_text(stmt, i);

int length = sqlite3_column_bytes(stmt, i);

row->push_back(new Values::Text(name, length, text));

} break;

case SQLITE_BLOB: {

const void* blob = sqlite3_column_blob(stmt, i);

int length = sqlite3_column_bytes(stmt, i);

row->push_back(new Values::Blob(name, length, blob));

} break;

case SQLITE_NULL: {

row->push_back(new Values::Null(name));

} break;

default:

assert(false);

}

}

NanScope();

Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());

OPTIONAL_ARGUMENT_FUNCTION(0, callback);

Baton* baton = new Baton(stmt, callback);

stmt->Schedule(Finalize, baton);

NanReturnValue(NanObjectWrapHandle(stmt->db));

baton->stmt->Finalize();

// Fire callback in case there was one.

Local<Function> cb = NanPersistentToLocal(baton->callback);

if (!cb.IsEmpty() && cb->IsFunction()) {

TRY_CATCH_CALL(NanObjectWrapHandle(baton->stmt), cb, 0, NULL);

}

delete baton;

assert(!finalized);

finalized = true;

CleanQueue();

// Finalize returns the status code of the last operation. We already fired

// error events in case those failed.

sqlite3_finalize(_handle);

_handle = NULL;

db->Unref();

if (prepared && !queue.empty()) {

// This statement has already been prepared and is now finalized.

// Fire error for all remaining items in the queue.

EXCEPTION(String::New("Statement is already finalized"), SQLITE_MISUSE, exception);

Local<Value> argv[] = { exception };

bool called = false;

// Clear out the queue so that this object can get GC'ed.

while (!queue.empty()) {

Call* call = queue.front();

queue.pop();

Local<Function> cb = NanPersistentToLocal(call->baton->callback);

if (prepared && !cb.IsEmpty() &&

cb->IsFunction()) {

TRY_CATCH_CALL(NanObjectWrapHandle(this), cb, 1, argv);

called = true;

}

// We don't call the actual callback, so we have to make sure that

// the baton gets destroyed.

delete call->baton;

delete call;

}

// When we couldn't call a callback function, emit an error on the

// Statement object.

if (!called) {

Local<Value> args[] = { String::NewSymbol("error"), exception };

EMIT_EVENT(NanObjectWrapHandle(this), 2, args);

}

}

else while (!queue.empty()) {

// Just delete all items in the queue; we already fired an event when

// preparing the statement failed.

Call* call = queue.front();

queue.pop();

// We don't call the actual callback, so we have to make sure that

// the baton gets destroyed.

delete call->baton;

delete call;

}