Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit e436fca

Browse filesBrowse files
kkaeferkkaefer
committed
implement Statement#each with ev_async
1 parent 09e6814 commit e436fca
Copy full SHA for e436fca

4 files changed

+170-15Lines changed: 170 additions & 15 deletions

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎lib/sqlite3.js‎

Copy file name to clipboardExpand all lines: lib/sqlite3.js
+12Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,19 @@ Database.prototype.all = function(sql) {
4242
return this;
4343
}
4444

45+
// Database#each(sql, [bind1, bind2, ...], [callback])
46+
Database.prototype.each = function(sql) {
47+
var statement = new Statement(this, sql);
48+
statement.each.apply(statement, Array.prototype.slice.call(arguments, 1)).finalize();
49+
return this;
50+
}
51+
4552
Database.prototype.execute = function() {
4653
console.warn('Database#execute() is deprecated. Use Database#all() instead.');
4754
return this.all.apply(this, arguments);
4855
};
56+
57+
Database.prototype.query = function() {
58+
console.warn('Database#query() is deprecated. Use Database#each() instead.');
59+
return this.each.apply(this, arguments);
60+
};
Collapse file

‎src/statement.cc‎

Copy file name to clipboardExpand all lines: src/statement.cc
+113-1Lines changed: 113 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ void Statement::Init(Handle<Object> target) {
2727
NODE_SET_PROTOTYPE_METHOD(constructor_template, "get", Get);
2828
NODE_SET_PROTOTYPE_METHOD(constructor_template, "run", Run);
2929
NODE_SET_PROTOTYPE_METHOD(constructor_template, "all", All);
30+
NODE_SET_PROTOTYPE_METHOD(constructor_template, "each", Each);
3031
NODE_SET_PROTOTYPE_METHOD(constructor_template, "reset", Reset);
3132
NODE_SET_PROTOTYPE_METHOD(constructor_template, "finalize", Finalize);
3233

@@ -101,14 +102,14 @@ Handle<Value> Statement::New(const Arguments& args) {
101102

102103
Statement* stmt = new Statement(db);
103104
stmt->Wrap(args.This());
105+
104106
PrepareBaton* baton = new PrepareBaton(db, Local<Function>::Cast(args[2]), stmt);
105107
baton->sql = std::string(*String::Utf8Value(sql));
106108
db->Schedule(EIO_BeginPrepare, baton);
107109

108110
return args.This();
109111
}
110112

111-
112113
void Statement::EIO_BeginPrepare(Database::Baton* baton) {
113114
assert(baton->db->open);
114115
baton->db->pending++;
@@ -520,6 +521,117 @@ int Statement::EIO_AfterAll(eio_req *req) {
520521
return 0;
521522
}
522523

524+
Handle<Value> Statement::Each(const Arguments& args) {
525+
HandleScope scope;
526+
Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());
527+
528+
Baton* baton = stmt->Bind<Baton>(args);
529+
if (baton == NULL) {
530+
return ThrowException(Exception::Error(String::New("Data type is not supported")));
531+
}
532+
else {
533+
stmt->Schedule(EIO_BeginEach, baton);
534+
return args.This();
535+
}
536+
}
537+
538+
void Statement::EIO_BeginEach(Baton* baton) {
539+
STATEMENT_BEGIN(Each);
540+
}
541+
542+
int Statement::EIO_Each(eio_req *req) {
543+
STATEMENT_INIT(Baton);
544+
545+
Async* async = new Async(stmt, baton->callback, AsyncEach);
546+
547+
sqlite3_mutex* mtx = sqlite3_db_mutex(stmt->db->handle);
548+
549+
int retrieved = 0;
550+
551+
// Make sure that we also reset when there are no parameters.
552+
if (!baton->parameters.size()) {
553+
sqlite3_reset(stmt->handle);
554+
}
555+
556+
if (stmt->Bind(baton->parameters)) {
557+
while (true) {
558+
sqlite3_mutex_enter(mtx);
559+
stmt->status = sqlite3_step(stmt->handle);
560+
if (stmt->status == SQLITE_ROW) {
561+
sqlite3_mutex_leave(mtx);
562+
Data::Row* row = new Data::Row();
563+
GetRow(row, stmt->handle);
564+
565+
pthread_mutex_lock(&async->mutex);
566+
async->data.push_back(row);
567+
retrieved++;
568+
pthread_mutex_unlock(&async->mutex);
569+
570+
ev_async_send(EV_DEFAULT_ &async->watcher);
571+
}
572+
else {
573+
if (stmt->status != SQLITE_DONE) {
574+
stmt->message = std::string(sqlite3_errmsg(stmt->db->handle));
575+
}
576+
sqlite3_mutex_leave(mtx);
577+
break;
578+
}
579+
}
580+
}
581+
582+
async->completed = true;
583+
ev_async_send(EV_DEFAULT_ &async->watcher);
584+
585+
return 0;
586+
}
587+
588+
void Statement::AsyncEach(EV_P_ ev_async *w, int revents) {
589+
HandleScope scope;
590+
Async* async = static_cast<Async*>(w->data);
591+
592+
while (true) {
593+
// Get the contents out of the data cache for us to process in the JS callback.
594+
Data::Rows rows;
595+
pthread_mutex_lock(&async->mutex);
596+
rows.swap(async->data);
597+
pthread_mutex_unlock(&async->mutex);
598+
599+
if (rows.empty()) {
600+
break;
601+
}
602+
603+
if (!async->callback.IsEmpty() && async->callback->IsFunction()) {
604+
Local<Value> argv[2];
605+
argv[0] = Local<Value>::New(Null());
606+
607+
Data::Rows::const_iterator it = rows.begin();
608+
Data::Rows::const_iterator end = rows.end();
609+
for (int i = 0; it < end; it++, i++) {
610+
argv[1] = RowToJS(*it);
611+
TRY_CATCH_CALL(async->stmt->handle_, async->callback, 2, argv);
612+
delete *it;
613+
}
614+
}
615+
}
616+
617+
if (async->completed) {
618+
delete async;
619+
w->data = NULL;
620+
}
621+
}
622+
623+
int Statement::EIO_AfterEach(eio_req *req) {
624+
HandleScope scope;
625+
STATEMENT_INIT(Baton);
626+
627+
if (stmt->status != SQLITE_DONE) {
628+
Error(baton);
629+
}
630+
631+
STATEMENT_END();
632+
return 0;
633+
}
634+
523635
Handle<Value> Statement::Reset(const Arguments& args) {
524636
HandleScope scope;
525637
Statement* stmt = ObjectWrap::Unwrap<Statement>(args.This());
Collapse file

‎src/statement.h‎

Copy file name to clipboardExpand all lines: src/statement.h
+37-6Lines changed: 37 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -117,13 +117,41 @@ class Statement : public EventEmitter {
117117
Baton* baton;
118118
};
119119

120+
typedef void (*Async_Callback)(EV_P_ ev_async *w, int revents);
121+
122+
struct Async {
123+
ev_async watcher;
124+
Statement* stmt;
125+
Data::Rows data;
126+
pthread_mutex_t mutex;
127+
Persistent<Function> callback;
128+
bool completed;
129+
130+
Async(Statement* st, Handle<Function> cb, Async_Callback async_cb) :
131+
stmt(st), completed(false) {
132+
watcher.data = this;
133+
ev_async_init(&watcher, async_cb);
134+
ev_async_start(EV_DEFAULT_UC_ &watcher);
135+
callback = Persistent<Function>::New(cb);
136+
stmt->Ref();
137+
pthread_mutex_init(&mutex, NULL);
138+
}
139+
140+
~Async() {
141+
callback.Dispose();
142+
stmt->Unref();
143+
pthread_mutex_destroy(&mutex);
144+
ev_async_stop(EV_DEFAULT_UC_ &watcher);
145+
}
146+
};
147+
120148
Statement(Database* db_) : EventEmitter(),
121-
db(db_),
122-
handle(NULL),
123-
status(SQLITE_OK),
124-
prepared(false),
125-
locked(false),
126-
finalized(false) {
149+
db(db_),
150+
handle(NULL),
151+
status(SQLITE_OK),
152+
prepared(false),
153+
locked(false),
154+
finalized(false) {
127155
db->Ref();
128156
}
129157

@@ -140,8 +168,11 @@ class Statement : public EventEmitter {
140168
EIO_DEFINITION(Get);
141169
EIO_DEFINITION(Run);
142170
EIO_DEFINITION(All);
171+
EIO_DEFINITION(Each);
143172
EIO_DEFINITION(Reset);
144173

174+
static void AsyncEach(EV_P_ ev_async *w, int revents);
175+
145176
static Handle<Value> Finalize(const Arguments& args);
146177
static void Finalize(Baton* baton);
147178
void Finalize();
Collapse file

‎test/unicode.test.js‎

Copy file name to clipboardExpand all lines: test/unicode.test.js
+8-8Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,17 @@
11
var sqlite3 = require('sqlite3');
22
var assert = require('assert');
33

4+
function randomString() {
5+
var str = '';
6+
for (var i = Math.random() * 300; i > 0; i--) {
7+
str += String.fromCharCode(Math.floor(Math.random() * 65536));
8+
}
9+
return str;
10+
};
11+
412
exports['test unicode characters'] = function(beforeExit) {
513
var db = new sqlite3.Database(':memory:');
614

7-
function randomString() {
8-
var str = '';
9-
for (var i = Math.random() * 300; i > 0; i--) {
10-
str += String.fromCharCode(Math.floor(Math.random() * 65536));
11-
}
12-
return str;
13-
};
14-
1515
// Generate random data.
1616
var data = [];
1717
var length = Math.floor(Math.random() * 1000) + 200;

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.