Brian M. Carlson
10 years ago
parent
b325971fdf
commit
667c528ea6
@ -1,38 +0,0 @@
|
||||
{
|
||||
'targets': [
|
||||
{
|
||||
'target_name': 'binding',
|
||||
'sources': ['src/binding.cc'],
|
||||
'include_dirs': [
|
||||
'<!@(pg_config --includedir)',
|
||||
'<!(node -e "require(\'nan\')")'
|
||||
],
|
||||
'conditions' : [
|
||||
['OS=="win"', {
|
||||
'conditions' : [
|
||||
['"<!@(cmd /C where /Q pg_config || echo n)"!="n"',
|
||||
{
|
||||
'libraries' : ['libpq.lib'],
|
||||
'msvs_settings': {
|
||||
'VCLinkerTool' : {
|
||||
'AdditionalLibraryDirectories' : [
|
||||
'<!@(pg_config --libdir)\\'
|
||||
]
|
||||
},
|
||||
}
|
||||
}
|
||||
]
|
||||
]
|
||||
}, { # OS!="win"
|
||||
'conditions' : [
|
||||
['"y"!="n"', # ToDo: add pg_config existance condition that works on linux
|
||||
{
|
||||
'libraries' : ['-lpq -L<!@(pg_config --libdir)']
|
||||
}
|
||||
]
|
||||
]
|
||||
}]
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
@ -1,953 +0,0 @@
|
||||
#include <pg_config.h>
|
||||
#include <libpq-fe.h>
|
||||
#include <nan.h>
|
||||
#include <string.h>
|
||||
#include <assert.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
#define LOG(msg) printf("%s\n",msg);
|
||||
#define TRACE(msg) //printf("%s\n", msg);
|
||||
|
||||
#if PG_VERSION_NUM >= 90000
|
||||
#define ESCAPE_SUPPORTED
|
||||
#endif
|
||||
|
||||
#if PG_VERSION_NUM >= 90200
|
||||
#define SINGLE_ROW_SUPPORTED
|
||||
#endif
|
||||
|
||||
#define THROW(msg) NanThrowError(msg); NanReturnUndefined();
|
||||
|
||||
using namespace v8;
|
||||
using namespace node;
|
||||
|
||||
class Connection : public ObjectWrap {
|
||||
|
||||
public:
|
||||
|
||||
//creates the V8 objects & attaches them to the module (target)
|
||||
static void
|
||||
Init (Handle<Object> target)
|
||||
{
|
||||
NanScope();
|
||||
Local<FunctionTemplate> t = NanNew<FunctionTemplate>(New);
|
||||
|
||||
t->InstanceTemplate()->SetInternalFieldCount(1);
|
||||
t->SetClassName(NanNew("Connection"));
|
||||
|
||||
NanSetPrototypeTemplate(t, "connect", NanNew<FunctionTemplate>(Connect));
|
||||
#ifdef ESCAPE_SUPPORTED
|
||||
NanSetPrototypeTemplate(t, "escapeIdentifier", NanNew<FunctionTemplate>(EscapeIdentifier));
|
||||
NanSetPrototypeTemplate(t, "escapeLiteral", NanNew<FunctionTemplate>(EscapeLiteral));
|
||||
#endif
|
||||
NanSetPrototypeTemplate(t, "_sendQuery", NanNew<FunctionTemplate>(SendQuery));
|
||||
NanSetPrototypeTemplate(t, "_sendQueryWithParams", NanNew<FunctionTemplate>(SendQueryWithParams));
|
||||
NanSetPrototypeTemplate(t, "_sendPrepare", NanNew<FunctionTemplate>(SendPrepare));
|
||||
NanSetPrototypeTemplate(t, "_sendQueryPrepared", NanNew<FunctionTemplate>(SendQueryPrepared));
|
||||
NanSetPrototypeTemplate(t, "cancel", NanNew<FunctionTemplate>(Cancel));
|
||||
NanSetPrototypeTemplate(t, "end", NanNew<FunctionTemplate>(End));
|
||||
NanSetPrototypeTemplate(t, "_sendCopyFromChunk", NanNew<FunctionTemplate>(SendCopyFromChunk));
|
||||
NanSetPrototypeTemplate(t, "_endCopyFrom", NanNew<FunctionTemplate>(EndCopyFrom));
|
||||
target->Set(NanNew("Connection"), t->GetFunction());
|
||||
TRACE("created class");
|
||||
}
|
||||
|
||||
//static function called by libuv as callback entrypoint
|
||||
static void
|
||||
io_event(uv_poll_t* w, int status, int revents)
|
||||
{
|
||||
|
||||
TRACE("Received IO event");
|
||||
|
||||
if(status == -1) {
|
||||
TRACE("Connection error. -1 status from lib_uv_poll");
|
||||
}
|
||||
|
||||
Connection *connection = static_cast<Connection*>(w->data);
|
||||
connection->HandleIOEvent(revents);
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#connect
|
||||
static NAN_METHOD(Connect)
|
||||
{
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
if(args.Length() == 0 || !args[0]->IsString()) {
|
||||
THROW("Must include connection string as only argument to connect");
|
||||
}
|
||||
|
||||
String::Utf8Value conninfo(args[0]->ToString());
|
||||
bool success = self->Connect(*conninfo);
|
||||
if(!success) {
|
||||
self -> EmitLastError();
|
||||
self -> DestroyConnection();
|
||||
}
|
||||
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#cancel
|
||||
static NAN_METHOD(Cancel)
|
||||
{
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
|
||||
bool success = self->Cancel();
|
||||
if(!success) {
|
||||
self -> EmitLastError();
|
||||
self -> DestroyConnection();
|
||||
}
|
||||
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
#ifdef ESCAPE_SUPPORTED
|
||||
//v8 entry point into Connection#escapeIdentifier
|
||||
static NAN_METHOD(EscapeIdentifier)
|
||||
{
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
|
||||
char* inputStr = MallocCString(args[0]);
|
||||
|
||||
if(!inputStr) {
|
||||
THROW("Unable to allocate memory for a string in EscapeIdentifier.")
|
||||
}
|
||||
|
||||
char* escapedStr = self->EscapeIdentifier(inputStr);
|
||||
free(inputStr);
|
||||
|
||||
if(escapedStr == NULL) {
|
||||
THROW(self->GetLastError());
|
||||
}
|
||||
|
||||
Local<Value> jsStr = NanNew<String>(escapedStr, strlen(escapedStr));
|
||||
PQfreemem(escapedStr);
|
||||
|
||||
NanReturnValue(jsStr);
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#escapeLiteral
|
||||
static NAN_METHOD(EscapeLiteral)
|
||||
{
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
|
||||
char* inputStr = MallocCString(args[0]);
|
||||
|
||||
if(!inputStr) {
|
||||
THROW("Unable to allocate memory for a string in EscapeIdentifier.")
|
||||
}
|
||||
|
||||
char* escapedStr = self->EscapeLiteral(inputStr);
|
||||
free(inputStr);
|
||||
|
||||
if(escapedStr == NULL) {
|
||||
THROW(self->GetLastError());
|
||||
}
|
||||
|
||||
Local<Value> jsStr = NanNew<String>(escapedStr, strlen(escapedStr));
|
||||
PQfreemem(escapedStr);
|
||||
|
||||
NanReturnValue(jsStr);
|
||||
}
|
||||
#endif
|
||||
|
||||
//v8 entry point into Connection#_sendQuery
|
||||
static NAN_METHOD(SendQuery)
|
||||
{
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
const char *lastErrorMessage;
|
||||
if(!args[0]->IsString()) {
|
||||
THROW("First parameter must be a string query");
|
||||
}
|
||||
|
||||
char* queryText = MallocCString(args[0]);
|
||||
bool singleRowMode = (bool)args[1]->Int32Value();
|
||||
|
||||
int result = self->Send(queryText, singleRowMode);
|
||||
free(queryText);
|
||||
if(result == 0) {
|
||||
lastErrorMessage = self->GetLastError();
|
||||
THROW(lastErrorMessage);
|
||||
}
|
||||
//TODO should we flush before throw?
|
||||
self->Flush();
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#_sendQueryWithParams
|
||||
static NAN_METHOD(SendQueryWithParams)
|
||||
{
|
||||
NanScope();
|
||||
//dispatch non-prepared parameterized query
|
||||
DispatchParameterizedQuery(args, false);
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#_sendPrepare(string queryName, string queryText, int nParams)
|
||||
static NAN_METHOD(SendPrepare)
|
||||
{
|
||||
NanScope();
|
||||
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
String::Utf8Value queryName(args[0]);
|
||||
String::Utf8Value queryText(args[1]);
|
||||
int length = args[2]->Int32Value();
|
||||
bool singleRowMode = (bool)args[3]->Int32Value();
|
||||
self->SendPrepare(*queryName, *queryText, length, singleRowMode);
|
||||
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#_sendQueryPrepared(string queryName, string[] paramValues)
|
||||
static NAN_METHOD(SendQueryPrepared)
|
||||
{
|
||||
NanScope();
|
||||
//dispatch prepared parameterized query
|
||||
DispatchParameterizedQuery(args, true);
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
static void DispatchParameterizedQuery(_NAN_METHOD_ARGS, bool isPrepared)
|
||||
{
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
|
||||
String::Utf8Value queryName(args[0]);
|
||||
//TODO this is much copy/pasta code
|
||||
if(!args[0]->IsString()) {
|
||||
NanThrowError("First parameter must be a string");
|
||||
return;
|
||||
}
|
||||
|
||||
if(!args[1]->IsArray()) {
|
||||
NanThrowError("Values must be an array");
|
||||
return;
|
||||
}
|
||||
|
||||
Local<Array> jsParams = Local<Array>::Cast(args[1]);
|
||||
int len = jsParams->Length();
|
||||
|
||||
|
||||
char** paramValues = ArgToCStringArray(jsParams);
|
||||
if(!paramValues) {
|
||||
NanThrowError("Unable to allocate char **paramValues from Local<Array> of v8 params");
|
||||
return;
|
||||
}
|
||||
|
||||
char* queryText = MallocCString(args[0]);
|
||||
bool singleRowMode = (bool)args[2]->Int32Value();
|
||||
|
||||
int result = 0;
|
||||
if(isPrepared) {
|
||||
result = self->SendPreparedQuery(queryText, len, paramValues, singleRowMode);
|
||||
} else {
|
||||
result = self->SendQueryParams(queryText, len, paramValues, singleRowMode);
|
||||
}
|
||||
|
||||
free(queryText);
|
||||
ReleaseCStringArray(paramValues, len);
|
||||
if(result == 1) {
|
||||
return;
|
||||
}
|
||||
self->EmitLastError();
|
||||
NanThrowError("Postgres returned non-1 result from query dispatch.");
|
||||
}
|
||||
|
||||
//v8 entry point into Connection#end
|
||||
static NAN_METHOD(End)
|
||||
{
|
||||
NanScope();
|
||||
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
|
||||
self->End();
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
uv_poll_t read_watcher_;
|
||||
uv_poll_t write_watcher_;
|
||||
PGconn *connection_;
|
||||
bool connecting_;
|
||||
bool ioInitialized_;
|
||||
bool copyOutMode_;
|
||||
bool copyInMode_;
|
||||
bool reading_;
|
||||
bool writing_;
|
||||
bool ended_;
|
||||
Connection () : ObjectWrap ()
|
||||
{
|
||||
connection_ = NULL;
|
||||
connecting_ = false;
|
||||
ioInitialized_ = false;
|
||||
copyOutMode_ = false;
|
||||
copyInMode_ = false;
|
||||
reading_ = false;
|
||||
writing_ = false;
|
||||
ended_ = false;
|
||||
TRACE("Initializing ev watchers");
|
||||
read_watcher_.data = this;
|
||||
write_watcher_.data = this;
|
||||
}
|
||||
|
||||
~Connection ()
|
||||
{
|
||||
}
|
||||
|
||||
static NAN_METHOD(SendCopyFromChunk) {
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
//TODO handle errors in some way
|
||||
if (args.Length() < 1 && !Buffer::HasInstance(args[0])) {
|
||||
THROW("SendCopyFromChunk requires 1 Buffer argument");
|
||||
}
|
||||
self->SendCopyFromChunk(args[0]->ToObject());
|
||||
NanReturnUndefined();
|
||||
}
|
||||
static NAN_METHOD(EndCopyFrom) {
|
||||
NanScope();
|
||||
Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
|
||||
char * error_msg = NULL;
|
||||
if (args[0]->IsString()) {
|
||||
error_msg = MallocCString(args[0]);
|
||||
}
|
||||
//TODO handle errors in some way
|
||||
self->EndCopyFrom(error_msg);
|
||||
free(error_msg);
|
||||
NanReturnUndefined();
|
||||
}
|
||||
|
||||
protected:
|
||||
//v8 entry point to constructor
|
||||
static NAN_METHOD(New)
|
||||
{
|
||||
NanScope();
|
||||
Connection *connection = new Connection();
|
||||
connection->Wrap(args.This());
|
||||
|
||||
NanReturnValue(args.This());
|
||||
}
|
||||
|
||||
#ifdef ESCAPE_SUPPORTED
|
||||
char * EscapeIdentifier(const char *str)
|
||||
{
|
||||
TRACE("js::EscapeIdentifier")
|
||||
return PQescapeIdentifier(connection_, str, strlen(str));
|
||||
}
|
||||
|
||||
char * EscapeLiteral(const char *str)
|
||||
{
|
||||
TRACE("js::EscapeLiteral")
|
||||
return PQescapeLiteral(connection_, str, strlen(str));
|
||||
}
|
||||
#endif
|
||||
|
||||
void enableSingleRowMode(bool enable)
|
||||
{
|
||||
#ifdef SINGLE_ROW_SUPPORTED
|
||||
if(enable == true) {
|
||||
int mode = PQsetSingleRowMode(connection_);
|
||||
if(mode == 1) {
|
||||
TRACE("PQsetSingleRowMode enabled")
|
||||
} else {
|
||||
TRACE("PQsetSingleRowMode disabled")
|
||||
}
|
||||
} else {
|
||||
TRACE("PQsetSingleRowMode disabled")
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
int Send(const char *queryText, bool singleRowMode)
|
||||
{
|
||||
TRACE("js::Send")
|
||||
int rv = PQsendQuery(connection_, queryText);
|
||||
enableSingleRowMode(singleRowMode);
|
||||
StartWrite();
|
||||
return rv;
|
||||
}
|
||||
|
||||
int SendQueryParams(const char *command, const int nParams, const char * const *paramValues, bool singleRowMode)
|
||||
{
|
||||
TRACE("js::SendQueryParams")
|
||||
int rv = PQsendQueryParams(connection_, command, nParams, NULL, paramValues, NULL, NULL, 0);
|
||||
enableSingleRowMode(singleRowMode);
|
||||
StartWrite();
|
||||
return rv;
|
||||
}
|
||||
|
||||
int SendPrepare(const char *name, const char *command, const int nParams, bool singleRowMode)
|
||||
{
|
||||
TRACE("js::SendPrepare")
|
||||
int rv = PQsendPrepare(connection_, name, command, nParams, NULL);
|
||||
enableSingleRowMode(singleRowMode);
|
||||
StartWrite();
|
||||
return rv;
|
||||
}
|
||||
|
||||
int SendPreparedQuery(const char *name, int nParams, const char * const *paramValues, bool singleRowMode)
|
||||
{
|
||||
int rv = PQsendQueryPrepared(connection_, name, nParams, paramValues, NULL, NULL, 0);
|
||||
enableSingleRowMode(singleRowMode);
|
||||
StartWrite();
|
||||
return rv;
|
||||
}
|
||||
|
||||
bool Cancel()
|
||||
{
|
||||
PGcancel* pgCancel = PQgetCancel(connection_);
|
||||
char errbuf[256];
|
||||
int result = PQcancel(pgCancel, errbuf, 256);
|
||||
StartWrite();
|
||||
PQfreeCancel(pgCancel);
|
||||
return result;
|
||||
}
|
||||
|
||||
//flushes socket
|
||||
void Flush()
|
||||
{
|
||||
if(PQflush(connection_) == 1) {
|
||||
TRACE("Flushing");
|
||||
uv_poll_start(&write_watcher_, UV_WRITABLE, io_event);
|
||||
}
|
||||
}
|
||||
|
||||
//safely destroys the connection at most 1 time
|
||||
void DestroyConnection()
|
||||
{
|
||||
if(connection_ != NULL) {
|
||||
PQfinish(connection_);
|
||||
connection_ = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
//initializes initial async connection to postgres via libpq
|
||||
//and hands off control to libev
|
||||
bool Connect(const char* conninfo)
|
||||
{
|
||||
if(ended_) return true;
|
||||
connection_ = PQconnectStart(conninfo);
|
||||
|
||||
if (!connection_) {
|
||||
LOG("Connection couldn't be created");
|
||||
}
|
||||
|
||||
ConnStatusType status = PQstatus(connection_);
|
||||
|
||||
if(CONNECTION_BAD == status) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (PQsetnonblocking(connection_, 1) == -1) {
|
||||
LOG("Unable to set connection to non-blocking");
|
||||
return false;
|
||||
}
|
||||
|
||||
int fd = PQsocket(connection_);
|
||||
if(fd < 0) {
|
||||
LOG("socket fd was negative. error");
|
||||
return false;
|
||||
}
|
||||
|
||||
assert(PQisnonblocking(connection_));
|
||||
|
||||
PQsetNoticeProcessor(connection_, NoticeReceiver, this);
|
||||
|
||||
TRACE("Setting watchers to socket");
|
||||
uv_poll_init(uv_default_loop(), &read_watcher_, fd);
|
||||
uv_poll_init(uv_default_loop(), &write_watcher_, fd);
|
||||
|
||||
ioInitialized_ = true;
|
||||
|
||||
connecting_ = true;
|
||||
StartWrite();
|
||||
|
||||
Ref();
|
||||
return true;
|
||||
}
|
||||
|
||||
static void NoticeReceiver(void *arg, const char *message)
|
||||
{
|
||||
Connection *self = (Connection*)arg;
|
||||
self->HandleNotice(message);
|
||||
}
|
||||
|
||||
void HandleNotice(const char *message)
|
||||
{
|
||||
NanScope();
|
||||
Handle<Value> notice = NanNew<String>(message);
|
||||
Emit("notice", ¬ice);
|
||||
}
|
||||
|
||||
//called to process io_events from libuv
|
||||
void HandleIOEvent(int revents)
|
||||
{
|
||||
|
||||
if(connecting_) {
|
||||
TRACE("Processing connecting_ io");
|
||||
HandleConnectionIO();
|
||||
return;
|
||||
}
|
||||
|
||||
if(revents & UV_READABLE) {
|
||||
TRACE("revents & UV_READABLE");
|
||||
TRACE("about to consume input");
|
||||
if(PQconsumeInput(connection_) == 0) {
|
||||
TRACE("could not read, terminating");
|
||||
End();
|
||||
EmitLastError();
|
||||
//LOG("Something happened, consume input is 0");
|
||||
return;
|
||||
}
|
||||
TRACE("Consumed");
|
||||
|
||||
//declare handlescope as this method is entered via a libuv callback
|
||||
//and not part of the public v8 interface
|
||||
NanScope();
|
||||
if (this->copyOutMode_) {
|
||||
this->HandleCopyOut();
|
||||
}
|
||||
if (!this->copyInMode_ && !this->copyOutMode_ && PQisBusy(connection_) == 0) {
|
||||
PGresult *result;
|
||||
bool didHandleResult = false;
|
||||
TRACE("PQgetResult");
|
||||
while ((result = PQgetResult(connection_))) {
|
||||
TRACE("HandleResult");
|
||||
didHandleResult = HandleResult(result);
|
||||
TRACE("PQClear");
|
||||
PQclear(result);
|
||||
if(!didHandleResult) {
|
||||
//this means that we are in copy in or copy out mode
|
||||
//in this situation PQgetResult will return same
|
||||
//result untill all data will be read (copy out) or
|
||||
//until data end notification (copy in)
|
||||
//and because of this, we need to break cycle
|
||||
break;
|
||||
}
|
||||
}
|
||||
//might have fired from notification
|
||||
if(didHandleResult) {
|
||||
Emit("_readyForQuery");
|
||||
}
|
||||
}
|
||||
|
||||
PGnotify *notify;
|
||||
TRACE("PQnotifies");
|
||||
while ((notify = PQnotifies(connection_))) {
|
||||
Local<Object> result = NanNew<Object>();
|
||||
result->Set(NanNew("channel"), NanNew(notify->relname));
|
||||
result->Set(NanNew("payload"), NanNew(notify->extra));
|
||||
Handle<Value> res = (Handle<Value>)result;
|
||||
Emit("notification", &res);
|
||||
PQfreemem(notify);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if(revents & UV_WRITABLE) {
|
||||
TRACE("revents & UV_WRITABLE");
|
||||
if (PQflush(connection_) == 0) {
|
||||
//nothing left to write, poll the socket for more to read
|
||||
StartRead();
|
||||
}
|
||||
}
|
||||
}
|
||||
bool HandleCopyOut () {
|
||||
char * buffer = NULL;
|
||||
int copied;
|
||||
copied = PQgetCopyData(connection_, &buffer, 1);
|
||||
while (copied > 0) {
|
||||
Local<Value> node_chunk = NanNewBufferHandle(buffer, copied);
|
||||
Emit("copyData", &node_chunk);
|
||||
PQfreemem(buffer);
|
||||
copied = PQgetCopyData(connection_, &buffer, 1);
|
||||
}
|
||||
if (copied == 0) {
|
||||
//wait for next read ready
|
||||
//result was not handled completely
|
||||
return false;
|
||||
} else if (copied == -1) {
|
||||
this->copyOutMode_ = false;
|
||||
return true;
|
||||
} else if (copied == -2) {
|
||||
this->copyOutMode_ = false;
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
//maps the postgres tuple results to v8 objects
|
||||
//and emits row events
|
||||
//TODO look at emitting fewer events because the back & forth between
|
||||
//javascript & c++ might introduce overhead (requires benchmarking)
|
||||
void EmitRowDescription(const PGresult* result)
|
||||
{
|
||||
NanScope();
|
||||
Local<Array> row = NanNew<Array>();
|
||||
int fieldCount = PQnfields(result);
|
||||
for(int fieldNumber = 0; fieldNumber < fieldCount; fieldNumber++) {
|
||||
Local<Object> field = NanNew<Object>();
|
||||
//name of field
|
||||
char* fieldName = PQfname(result, fieldNumber);
|
||||
field->Set(NanNew("name"), NanNew(fieldName));
|
||||
|
||||
//oid of type of field
|
||||
int fieldType = PQftype(result, fieldNumber);
|
||||
field->Set(NanNew("dataTypeID"), NanNew(fieldType));
|
||||
|
||||
row->Set(NanNew(fieldNumber), field);
|
||||
}
|
||||
|
||||
Handle<Value> e = (Handle<Value>)row;
|
||||
Emit("_rowDescription", &e);
|
||||
}
|
||||
|
||||
bool HandleResult(PGresult* result)
|
||||
{
|
||||
TRACE("PQresultStatus");
|
||||
ExecStatusType status = PQresultStatus(result);
|
||||
switch(status) {
|
||||
case PGRES_TUPLES_OK:
|
||||
#ifdef SINGLE_ROW_SUPPORTED
|
||||
case PGRES_SINGLE_TUPLE:
|
||||
#endif
|
||||
{
|
||||
EmitRowDescription(result);
|
||||
HandleTuplesResult(result);
|
||||
EmitCommandMetaData(result);
|
||||
return true;
|
||||
}
|
||||
break;
|
||||
case PGRES_FATAL_ERROR:
|
||||
{
|
||||
TRACE("HandleErrorResult");
|
||||
HandleErrorResult(result);
|
||||
return true;
|
||||
}
|
||||
break;
|
||||
case PGRES_COMMAND_OK:
|
||||
case PGRES_EMPTY_QUERY:
|
||||
{
|
||||
EmitCommandMetaData(result);
|
||||
return true;
|
||||
}
|
||||
break;
|
||||
case PGRES_COPY_IN:
|
||||
{
|
||||
this->copyInMode_ = true;
|
||||
Emit("copyInResponse");
|
||||
return false;
|
||||
}
|
||||
break;
|
||||
case PGRES_COPY_OUT:
|
||||
{
|
||||
this->copyOutMode_ = true;
|
||||
Emit("copyOutResponse");
|
||||
return this->HandleCopyOut();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
printf("YOU SHOULD NEVER SEE THIS! PLEASE OPEN AN ISSUE ON GITHUB! Unrecogized query status: %s\n", PQresStatus(status));
|
||||
break;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
void EmitCommandMetaData(PGresult* result)
|
||||
{
|
||||
NanScope();
|
||||
Local<Object> info = NanNew<Object>();
|
||||
info->Set(NanNew("command"), NanNew(PQcmdStatus(result)));
|
||||
info->Set(NanNew("value"), NanNew(PQcmdTuples(result)));
|
||||
Handle<Value> e = (Handle<Value>)info;
|
||||
Emit("_cmdStatus", &e);
|
||||
}
|
||||
|
||||
//maps the postgres tuple results to v8 objects
|
||||
//and emits row events
|
||||
//TODO look at emitting fewer events because the back & forth between
|
||||
//javascript & c++ might introduce overhead (requires benchmarking)
|
||||
void HandleTuplesResult(const PGresult* result)
|
||||
{
|
||||
NanScope();
|
||||
int rowCount = PQntuples(result);
|
||||
for(int rowNumber = 0; rowNumber < rowCount; rowNumber++) {
|
||||
//create result object for this row
|
||||
Local<Array> row = NanNew<Array>();
|
||||
int fieldCount = PQnfields(result);
|
||||
for(int fieldNumber = 0; fieldNumber < fieldCount; fieldNumber++) {
|
||||
|
||||
//value of field
|
||||
if(PQgetisnull(result, rowNumber, fieldNumber)) {
|
||||
row->Set(NanNew(fieldNumber), NanNull());
|
||||
} else {
|
||||
char* fieldValue = PQgetvalue(result, rowNumber, fieldNumber);
|
||||
row->Set(NanNew(fieldNumber), NanNew(fieldValue));
|
||||
}
|
||||
}
|
||||
|
||||
Handle<Value> e = (Handle<Value>)row;
|
||||
Emit("_row", &e);
|
||||
}
|
||||
}
|
||||
|
||||
void HandleErrorResult(const PGresult* result)
|
||||
{
|
||||
NanScope();
|
||||
//instantiate the return object as an Error with the summary Postgres message
|
||||
TRACE("ReadResultField");
|
||||
const char* errorMessage = PQresultErrorField(result, PG_DIAG_MESSAGE_PRIMARY);
|
||||
if(!errorMessage) {
|
||||
//there is no error, it has already been consumed in the last
|
||||
//read-loop callback
|
||||
return;
|
||||
}
|
||||
Local<Object> msg = Local<Object>::Cast(NanError(errorMessage));
|
||||
TRACE("AttachErrorFields");
|
||||
//add the other information returned by Postgres to the error object
|
||||
AttachErrorField(result, msg, NanNew("severity"), PG_DIAG_SEVERITY);
|
||||
AttachErrorField(result, msg, NanNew("code"), PG_DIAG_SQLSTATE);
|
||||
AttachErrorField(result, msg, NanNew("detail"), PG_DIAG_MESSAGE_DETAIL);
|
||||
AttachErrorField(result, msg, NanNew("hint"), PG_DIAG_MESSAGE_HINT);
|
||||
AttachErrorField(result, msg, NanNew("position"), PG_DIAG_STATEMENT_POSITION);
|
||||
AttachErrorField(result, msg, NanNew("internalPosition"), PG_DIAG_INTERNAL_POSITION);
|
||||
AttachErrorField(result, msg, NanNew("internalQuery"), PG_DIAG_INTERNAL_QUERY);
|
||||
AttachErrorField(result, msg, NanNew("where"), PG_DIAG_CONTEXT);
|
||||
AttachErrorField(result, msg, NanNew("file"), PG_DIAG_SOURCE_FILE);
|
||||
AttachErrorField(result, msg, NanNew("line"), PG_DIAG_SOURCE_LINE);
|
||||
AttachErrorField(result, msg, NanNew("routine"), PG_DIAG_SOURCE_FUNCTION);
|
||||
Handle<Value> m = msg;
|
||||
TRACE("EmitError");
|
||||
Emit("_error", &m);
|
||||
}
|
||||
|
||||
void AttachErrorField(const PGresult *result, const Local<Object> msg, const Local<String> symbol, int fieldcode)
|
||||
{
|
||||
NanScope();
|
||||
char *val = PQresultErrorField(result, fieldcode);
|
||||
if(val) {
|
||||
msg->Set(symbol, NanNew(val));
|
||||
}
|
||||
}
|
||||
|
||||
void End()
|
||||
{
|
||||
TRACE("stopping read & write");
|
||||
StopRead();
|
||||
StopWrite();
|
||||
DestroyConnection();
|
||||
Emit("_end");
|
||||
ended_ = true;
|
||||
}
|
||||
|
||||
private:
|
||||
//EventEmitter was removed from c++ in node v0.5.x
|
||||
void Emit(const char* message) {
|
||||
NanScope();
|
||||
Handle<Value> args[1] = { NanNew(message) };
|
||||
Emit(1, args);
|
||||
}
|
||||
|
||||
void Emit(const char* message, Handle<Value>* arg) {
|
||||
NanScope();
|
||||
Handle<Value> args[2] = { NanNew(message), *arg };
|
||||
Emit(2, args);
|
||||
}
|
||||
|
||||
void Emit(int length, Handle<Value> *args) {
|
||||
NanScope();
|
||||
|
||||
Local<Value> emit_v = NanObjectWrapHandle(this)->Get(NanNew("emit"));
|
||||
assert(emit_v->IsFunction());
|
||||
Local<Function> emit_f = emit_v.As<Function>();
|
||||
|
||||
TryCatch tc;
|
||||
emit_f->Call(NanObjectWrapHandle(this), length, args);
|
||||
if(tc.HasCaught()) {
|
||||
FatalException(tc);
|
||||
}
|
||||
}
|
||||
|
||||
void HandleConnectionIO()
|
||||
{
|
||||
PostgresPollingStatusType status = PQconnectPoll(connection_);
|
||||
switch(status) {
|
||||
case PGRES_POLLING_READING:
|
||||
TRACE("Polled: PGRES_POLLING_READING");
|
||||
StartRead();
|
||||
break;
|
||||
case PGRES_POLLING_WRITING:
|
||||
TRACE("Polled: PGRES_POLLING_WRITING");
|
||||
StartWrite();
|
||||
break;
|
||||
case PGRES_POLLING_FAILED:
|
||||
StopRead();
|
||||
StopWrite();
|
||||
TRACE("Polled: PGRES_POLLING_FAILED");
|
||||
EmitLastError();
|
||||
break;
|
||||
case PGRES_POLLING_OK:
|
||||
TRACE("Polled: PGRES_POLLING_OK");
|
||||
connecting_ = false;
|
||||
StartRead();
|
||||
Emit("connect");
|
||||
default:
|
||||
//printf("Unknown polling status: %d\n", status);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void EmitError(const char *message)
|
||||
{
|
||||
NanScope();
|
||||
Local<Value> exception = NanError(message);
|
||||
Emit("_error", &exception);
|
||||
}
|
||||
|
||||
void EmitLastError()
|
||||
{
|
||||
EmitError(PQerrorMessage(connection_));
|
||||
}
|
||||
|
||||
const char *GetLastError()
|
||||
{
|
||||
return PQerrorMessage(connection_);
|
||||
}
|
||||
|
||||
void StopWrite()
|
||||
{
|
||||
TRACE("write STOP");
|
||||
if(ioInitialized_ && writing_) {
|
||||
uv_poll_stop(&write_watcher_);
|
||||
writing_ = false;
|
||||
}
|
||||
}
|
||||
|
||||
void StartWrite()
|
||||
{
|
||||
TRACE("write START");
|
||||
if(reading_) {
|
||||
TRACE("stop READ to start WRITE");
|
||||
StopRead();
|
||||
}
|
||||
uv_poll_start(&write_watcher_, UV_WRITABLE, io_event);
|
||||
writing_ = true;
|
||||
}
|
||||
|
||||
void StopRead()
|
||||
{
|
||||
TRACE("read STOP");
|
||||
if(ioInitialized_ && reading_) {
|
||||
uv_poll_stop(&read_watcher_);
|
||||
reading_ = false;
|
||||
}
|
||||
}
|
||||
|
||||
void StartRead()
|
||||
{
|
||||
TRACE("read START");
|
||||
if(writing_) {
|
||||
TRACE("stop WRITE to start READ");
|
||||
StopWrite();
|
||||
}
|
||||
uv_poll_start(&read_watcher_, UV_READABLE, io_event);
|
||||
reading_ = true;
|
||||
}
|
||||
//Converts a v8 array to an array of cstrings
|
||||
//the result char** array must be free() when it is no longer needed
|
||||
//if for any reason the array cannot be created, returns 0
|
||||
static char** ArgToCStringArray(Local<Array> params)
|
||||
{
|
||||
int len = params->Length();
|
||||
char** paramValues = new char*[len];
|
||||
for(int i = 0; i < len; i++) {
|
||||
Handle<Value> val = params->Get(i);
|
||||
if(val->IsString()) {
|
||||
char* cString = MallocCString(val);
|
||||
//will be 0 if could not malloc
|
||||
if(!cString) {
|
||||
LOG("ArgToCStringArray: OUT OF MEMORY OR SOMETHING BAD!");
|
||||
ReleaseCStringArray(paramValues, i-1);
|
||||
return 0;
|
||||
}
|
||||
paramValues[i] = cString;
|
||||
} else if(val->IsNull()) {
|
||||
paramValues[i] = NULL;
|
||||
} else if(val->IsObject() && Buffer::HasInstance(val)) {
|
||||
char *cHexString = MallocCHexString(val->ToObject());
|
||||
if(!cHexString) {
|
||||
LOG("ArgToCStringArray: OUT OF MEMORY OR SOMETHING BAD!");
|
||||
ReleaseCStringArray(paramValues, i-1);
|
||||
return 0;
|
||||
}
|
||||
paramValues[i] = cHexString;
|
||||
} else {
|
||||
//a paramter was not a string
|
||||
LOG("Parameter not a string or buffer");
|
||||
ReleaseCStringArray(paramValues, i-1);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
return paramValues;
|
||||
}
|
||||
|
||||
//helper function to release cString arrays
|
||||
static void ReleaseCStringArray(char **strArray, int len)
|
||||
{
|
||||
for(int i = 0; i < len; i++) {
|
||||
free(strArray[i]);
|
||||
}
|
||||
delete [] strArray;
|
||||
}
|
||||
|
||||
//helper function to malloc new string from v8string
|
||||
static char* MallocCString(v8::Handle<Value> v8String)
|
||||
{
|
||||
String::Utf8Value utf8String(v8String->ToString());
|
||||
char *cString = (char *) malloc(strlen(*utf8String) + 1);
|
||||
if(!cString) {
|
||||
return cString;
|
||||
}
|
||||
strcpy(cString, *utf8String);
|
||||
return cString;
|
||||
}
|
||||
|
||||
//helper function to Malloc a Bytea encoded Hex string from a buffer
|
||||
static char* MallocCHexString(v8::Handle<Object> buf)
|
||||
{
|
||||
char* bufferData = Buffer::Data(buf);
|
||||
size_t hexStringLen = Buffer::Length(buf)*2 + 3;
|
||||
char *cHexString = (char *) malloc(hexStringLen);
|
||||
if(!cHexString) {
|
||||
return cHexString;
|
||||
}
|
||||
strcpy(cHexString, "\\x");
|
||||
for (uint32_t i = 0, k = 2; k < hexStringLen; i += 1, k += 2) {
|
||||
static const char hex[] = "0123456789abcdef";
|
||||
uint8_t val = static_cast<uint8_t>(bufferData[i]);
|
||||
cHexString[k + 0] = hex[val >> 4];
|
||||
cHexString[k + 1] = hex[val & 15];
|
||||
}
|
||||
cHexString[hexStringLen-1] = 0;
|
||||
return cHexString;
|
||||
}
|
||||
|
||||
void SendCopyFromChunk(Handle<Object> chunk) {
|
||||
PQputCopyData(connection_, Buffer::Data(chunk), Buffer::Length(chunk));
|
||||
}
|
||||
void EndCopyFrom(char * error_msg) {
|
||||
PQputCopyEnd(connection_, error_msg);
|
||||
this->copyInMode_ = false;
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
|
||||
extern "C" void init (Handle<Object> target)
|
||||
{
|
||||
NanScope();
|
||||
Connection::Init(target);
|
||||
}
|
||||
NODE_MODULE(binding, init)
|
||||
@ -1,168 +0,0 @@
|
||||
var helper = require(__dirname + '/../test-helper');
|
||||
var pg = require(__dirname + '/../../../lib');
|
||||
if(helper.args.native) {
|
||||
pg = require(__dirname + '/../../../lib').native;
|
||||
}
|
||||
var ROWS_TO_INSERT = 1000;
|
||||
var prepareTable = function (client, callback) {
|
||||
client.query(
|
||||
'CREATE TEMP TABLE copy_test (id SERIAL, name CHARACTER VARYING(10), age INT)',
|
||||
assert.calls(function (err, result) {
|
||||
assert.equal(err, null,
|
||||
err && err.message ? "create table query should not fail: " + err.message : null);
|
||||
callback();
|
||||
})
|
||||
);
|
||||
};
|
||||
test('COPY FROM', function () {
|
||||
pg.connect(helper.config, function (error, client, done) {
|
||||
assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
|
||||
prepareTable(client, function () {
|
||||
var stream = client.copyFrom("COPY copy_test (name, age) FROM stdin WITH CSV");
|
||||
stream.on('error', function (error) {
|
||||
assert.ok(false, "COPY FROM stream should not emit errors" + helper.sys.inspect(error));
|
||||
});
|
||||
for (var i = 0; i < ROWS_TO_INSERT; i++) {
|
||||
stream.write( String(Date.now() + Math.random()).slice(0,10) + ',' + i + '\n');
|
||||
}
|
||||
assert.emits(stream, 'close', function () {
|
||||
client.query("SELECT count(*), sum(age) from copy_test", function (err, result) {
|
||||
assert.equal(err, null, "Query should not fail");
|
||||
assert.lengthIs(result.rows, 1)
|
||||
assert.equal(result.rows[0].sum, ROWS_TO_INSERT * (0 + ROWS_TO_INSERT -1)/2);
|
||||
assert.equal(result.rows[0].count, ROWS_TO_INSERT);
|
||||
done();
|
||||
});
|
||||
}, "COPY FROM stream should emit close after query end");
|
||||
stream.end();
|
||||
});
|
||||
});
|
||||
});
|
||||
test('COPY TO', function () {
|
||||
pg.connect(helper.config, function (error, client, done) {
|
||||
assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
|
||||
prepareTable(client, function () {
|
||||
var stream = client.copyTo("COPY person (id, name, age) TO stdin WITH CSV");
|
||||
var buf = new Buffer(0);
|
||||
stream.on('error', function (error) {
|
||||
assert.ok(false, "COPY TO stream should not emit errors" + helper.sys.inspect(error));
|
||||
});
|
||||
assert.emits(stream, 'data', function (chunk) {
|
||||
buf = Buffer.concat([buf, chunk]);
|
||||
}, "COPY IN stream should emit data event for each row");
|
||||
assert.emits(stream, 'end', function () {
|
||||
var lines = buf.toString().split('\n');
|
||||
assert.equal(lines.length >= 0, true, "copy in should return rows saved by copy from");
|
||||
assert.equal(lines[0].split(',').length, 3, "each line should consists of 3 fields");
|
||||
done();
|
||||
}, "COPY IN stream should emit end event after all rows");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
test('COPY TO, queue queries', function () {
|
||||
if(helper.config.native) return false;
|
||||
pg.connect(helper.config, assert.calls(function (error, client, done) {
|
||||
assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
|
||||
prepareTable(client, function () {
|
||||
var query1Done = false,
|
||||
copyQueryDone = false,
|
||||
query2Done = false;
|
||||
client.query("SELECT count(*) from person", function () {
|
||||
query1Done = true;
|
||||
assert.ok(!copyQueryDone && ! query2Done, "first query has to be executed before others");
|
||||
});
|
||||
var stream = client.copyTo("COPY person (id, name, age) TO stdin WITH CSV");
|
||||
//imitate long query, to make impossible,
|
||||
//that copy query end callback runs after
|
||||
//second query callback
|
||||
client.query("SELECT pg_sleep(1)", function () {
|
||||
query2Done = true;
|
||||
assert.ok(copyQueryDone && query2Done, "second query has to be executed after others");
|
||||
});
|
||||
var buf = new Buffer(0);
|
||||
stream.on('error', function (error) {
|
||||
assert.ok(false, "COPY TO stream should not emit errors" + helper.sys.inspect(error));
|
||||
});
|
||||
assert.emits(stream, 'data', function (chunk) {
|
||||
buf = Buffer.concat([buf, chunk]);
|
||||
}, "COPY IN stream should emit data event for each row");
|
||||
assert.emits(stream, 'end', function () {
|
||||
copyQueryDone = true;
|
||||
assert.ok(query1Done && ! query2Done, "copy query has to be executed before second query and after first");
|
||||
var lines = buf.toString().split('\n');
|
||||
assert.equal(lines.length >= 0, true, "copy in should return rows saved by copy from");
|
||||
assert.equal(lines[0].split(',').length, 3, "each line should consists of 3 fields");
|
||||
done();
|
||||
}, "COPY IN stream should emit end event after all rows");
|
||||
});
|
||||
}));
|
||||
});
|
||||
|
||||
test("COPY TO incorrect usage with large data", function () {
|
||||
if(helper.config.native) return false;
|
||||
//when many data is loaded from database (and it takes a lot of time)
|
||||
//there are chance, that query will be canceled before it ends
|
||||
//but if there are not so much data, cancel message may be
|
||||
//send after copy query ends
|
||||
//so we need to test both situations
|
||||
pg.connect(helper.config, assert.calls(function (error, client, done) {
|
||||
assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
|
||||
//intentionally incorrect usage of copy.
|
||||
//this has to report error in standart way, instead of just throwing exception
|
||||
client.query(
|
||||
"COPY (SELECT GENERATE_SERIES(1, 10000000)) TO STDOUT WITH CSV",
|
||||
assert.calls(function (error) {
|
||||
assert.ok(error, "error should be reported when sending copy to query with query method");
|
||||
client.query("SELECT 1", assert.calls(function (error, result) {
|
||||
assert.isNull(error, "incorrect copy usage should not break connection");
|
||||
assert.ok(result, "incorrect copy usage should not break connection");
|
||||
done();
|
||||
}));
|
||||
})
|
||||
);
|
||||
}));
|
||||
});
|
||||
|
||||
test("COPY TO incorrect usage with small data", function () {
|
||||
if(helper.config.native) return false;
|
||||
pg.connect(helper.config, assert.calls(function (error, client, done) {
|
||||
assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
|
||||
//intentionally incorrect usage of copy.
|
||||
//this has to report error in standart way, instead of just throwing exception
|
||||
client.query(
|
||||
"COPY (SELECT GENERATE_SERIES(1, 1)) TO STDOUT WITH CSV",
|
||||
assert.calls(function (error) {
|
||||
assert.ok(error, "error should be reported when sending copy to query with query method");
|
||||
client.query("SELECT 1", assert.calls(function (error, result) {
|
||||
assert.isNull(error, "incorrect copy usage should not break connection: " + error);
|
||||
assert.ok(result, "incorrect copy usage should not break connection");
|
||||
done();
|
||||
}));
|
||||
})
|
||||
);
|
||||
}));
|
||||
});
|
||||
|
||||
test("COPY FROM incorrect usage", function () {
|
||||
pg.connect(helper.config, function (error, client, done) {
|
||||
assert.equal(error, null, "Failed to connect: " + helper.sys.inspect(error));
|
||||
prepareTable(client, function () {
|
||||
//intentionally incorrect usage of copy.
|
||||
//this has to report error in standart way, instead of just throwing exception
|
||||
client.query(
|
||||
"COPY copy_test from STDIN WITH CSV",
|
||||
assert.calls(function (error) {
|
||||
assert.ok(error, "error should be reported when sending copy to query with query method");
|
||||
client.query("SELECT 1", assert.calls(function (error, result) {
|
||||
assert.isNull(error, "incorrect copy usage should not break connection: " + error);
|
||||
assert.ok(result, "incorrect copy usage should not break connection");
|
||||
done();
|
||||
pg.end(helper.config);
|
||||
}));
|
||||
})
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@ -1,36 +0,0 @@
|
||||
return console.log('these tests leak pg internals and are not helpful');
|
||||
var helper = require(__dirname+"/../test-helper");
|
||||
var Client = require(__dirname + "/../../lib/native");
|
||||
test('COPY FROM events check', function () {
|
||||
var con = new Client(helper.config);
|
||||
var stdinStream = con.copyFrom('COPY person FROM STDIN');
|
||||
|
||||
assert.emits(con, 'copyInResponse', function () { stdinStream.end(); },
|
||||
"backend should emit copyInResponse after COPY FROM query");
|
||||
|
||||
assert.emits(con, '_readyForQuery', function () { con.end(); },
|
||||
"backend should emit _readyForQuery after data will be coped to stdin stream");
|
||||
con.connect();
|
||||
});
|
||||
|
||||
test('COPY TO events check', function () {
|
||||
var con = new Client(helper.config),
|
||||
stdoutStream = con.copyTo('COPY person TO STDOUT');
|
||||
assert.emits(con, 'copyOutResponse',
|
||||
function () {},
|
||||
"backend should emit copyOutResponse on copyOutResponse message from server"
|
||||
);
|
||||
assert.emits(con, 'copyData',
|
||||
function () {
|
||||
},
|
||||
"backend should emit copyData on every data row"
|
||||
);
|
||||
assert.emits(con, '_readyForQuery',
|
||||
function () {
|
||||
con.end();
|
||||
},
|
||||
"backend should emit _readyForQuery after data will be coped to stdout stream"
|
||||
);
|
||||
con.connect();
|
||||
});
|
||||
|
||||
@ -1,23 +0,0 @@
|
||||
var helper = require(__dirname+"/../test-helper");
|
||||
var Client = require(__dirname + "/../../lib/native");
|
||||
test("COPY TO large amount of data from postgres", function () {
|
||||
//there were a bug in native implementation of COPY TO:
|
||||
//if there were too much data (if we face situation
|
||||
//when data is not ready while calling PQgetCopyData);
|
||||
//while loop in Connection::HandleIOEvent becomes infinite
|
||||
//in such way hanging node, consumes 100% cpu, and making connection unusable
|
||||
var con = new Client(helper.config),
|
||||
rowCount = 100000,
|
||||
stdoutStream = con.copyTo('COPY (select generate_series(1, ' + rowCount + ')) TO STDOUT');
|
||||
stdoutStream.on('data', function () {
|
||||
rowCount--;
|
||||
});
|
||||
stdoutStream.on('end', function () {
|
||||
assert.equal(rowCount, 0, "copy to should load exactly requested number of rows");
|
||||
con.query("SELECT 1", assert.calls(function (error, result) {
|
||||
assert.ok(!error && result, "loading large amount of data by copy to should not break connection");
|
||||
con.end();
|
||||
}));
|
||||
});
|
||||
con.connect();
|
||||
});
|
||||
@ -1,32 +0,0 @@
|
||||
import Options, Utils
|
||||
from os import unlink, symlink, popen
|
||||
from os.path import exists
|
||||
|
||||
srcdir = '.'
|
||||
blddir = 'build'
|
||||
VERSION = '0.0.1'
|
||||
|
||||
def set_options(opt):
|
||||
opt.tool_options('compiler_cxx')
|
||||
|
||||
def configure(conf):
|
||||
conf.check_tool('compiler_cxx')
|
||||
conf.check_tool('node_addon')
|
||||
|
||||
pg_config = conf.find_program('pg_config', var='PG_CONFIG', mandatory=True)
|
||||
pg_libdir = popen("%s --libdir" % pg_config).readline().strip()
|
||||
conf.env.append_value("LIBPATH_PG", pg_libdir)
|
||||
conf.env.append_value("LIB_PG", "pq")
|
||||
pg_includedir = popen("%s --includedir" % pg_config).readline().strip()
|
||||
conf.env.append_value("CPPPATH_PG", pg_includedir)
|
||||
|
||||
def build(bld):
|
||||
obj = bld.new_task_gen('cxx', 'shlib', 'node_addon')
|
||||
obj.cxxflags = ["-g", "-D_LARGEFILE_SOURCE", "-Wall"]
|
||||
obj.target = 'binding'
|
||||
obj.source = "./src/binding.cc"
|
||||
obj.uselib = "PG"
|
||||
|
||||
def test(test):
|
||||
Utils.exec_command("node test/native/connection-tests.js")
|
||||
Utils.exec_command("node test/native/evented-api-tests.js")
|
||||
Loading…
Reference in new issue