/* * This file is part of lem-postgres. * Copyright 2011 Emil Renner Berthing * * lem-postgres is free software: you can redistribute it and/or * modify it under the terms of the GNU General Public License as * published by the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * lem-postgres is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with lem-postgres. If not, see . */ #include #include #include #include struct connection { struct ev_io w; PGconn *conn; lua_State *T; }; static void push_error(lua_State *T, PGconn *conn) { const char *msg = PQerrorMessage(conn); const char *p; for (p = msg; *p != '\n' && *p != '\0'; p++); lua_pushnil(T); if (p > msg) lua_pushlstring(T, msg, p - msg); else lua_pushliteral(T, "unknown error"); } static int connection_gc(lua_State *T) { struct connection *c = lua_touserdata(T, 1); if (c->conn == NULL) return 0; ev_io_stop(EV_G_ &c->w); PQfinish(c->conn); return 0; } static int connection_close(lua_State *T) { struct connection *c; luaL_checktype(T, 1, LUA_TUSERDATA); c = lua_touserdata(T, 1); if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "already closed"); return 2; } if (c->T != NULL) { lua_pushnil(c->T); lua_pushliteral(c->T, "interrupted"); lem_queue(c->T, 2); c->T = NULL; } ev_io_stop(EV_G_ &c->w); PQfinish(c->conn); c->conn = NULL; lua_pushboolean(T, 1); return 1; } static void connect_handler(EV_P_ struct ev_io *w, int revents) { struct connection *c = (struct connection *)w; (void)revents; ev_io_stop(EV_A_ &c->w); switch (PQconnectPoll(c->conn)) { case PGRES_POLLING_READING: lem_debug("PGRES_POLLING_READING, socket = %d", PQsocket(c->conn)); ev_io_set(&c->w, PQsocket(c->conn), EV_READ); break; case PGRES_POLLING_WRITING: lem_debug("PGRES_POLLING_WRITING, socket = %d", PQsocket(c->conn)); ev_io_set(&c->w, PQsocket(c->conn), EV_WRITE); break; case PGRES_POLLING_FAILED: lem_debug("PGRES_POLLING_FAILED"); lua_settop(c->T, 0); push_error(c->T, c->conn); PQfinish(c->conn); c->conn = NULL; lem_queue(c->T, 2); c->T = NULL; return; case PGRES_POLLING_OK: lem_debug("PGRES_POLLING_OK"); lem_queue(c->T, 1); c->T = NULL; return; #ifndef NDEBUG case PGRES_POLLING_ACTIVE: assert(0); #endif } ev_io_start(EV_A_ &c->w); } static int connection_connect(lua_State *T) { const char *conninfo = luaL_checkstring(T, 1); PGconn *conn; ConnStatusType status; struct connection *c; conn = PQconnectStart(conninfo); if (conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "out of memory"); return 2; } status = PQstatus(conn); if (status == CONNECTION_BAD) { lem_debug("CONNECTION_BAD"); goto error; } c = lua_newuserdata(T, sizeof(struct connection)); lua_pushvalue(T, lua_upvalueindex(1)); lua_setmetatable(T, -2); c->conn = conn; switch (PQconnectPoll(conn)) { case PGRES_POLLING_READING: lem_debug("PGRES_POLLING_READING"); ev_io_init(&c->w, connect_handler, PQsocket(conn), EV_READ); break; case PGRES_POLLING_WRITING: lem_debug("PGRES_POLLING_WRITING"); ev_io_init(&c->w, connect_handler, PQsocket(conn), EV_WRITE); break; case PGRES_POLLING_FAILED: lem_debug("PGRES_POLLING_FAILED"); goto error; case PGRES_POLLING_OK: lem_debug("PGRES_POLLING_OK"); c->T = NULL; return 1; #ifndef NDEBUG case PGRES_POLLING_ACTIVE: assert(0); #endif } c->T = T; ev_io_start(EV_G_ &c->w); lua_replace(T, 1); lua_settop(T, 1); return lua_yield(T, 1); error: push_error(T, conn); PQfinish(conn); return 2; } static int connection_reset(lua_State *T) { struct connection *c; luaL_checktype(T, 1, LUA_TUSERDATA); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } if (PQresetStart(c->conn) != 1) goto error; c->w.cb = connect_handler; switch (PQconnectPoll(c->conn)) { case PGRES_POLLING_READING: lem_debug("PGRES_POLLING_READING"); ev_io_set(&c->w, PQsocket(c->conn), EV_READ); break; case PGRES_POLLING_WRITING: lem_debug("PGRES_POLLING_WRITING"); ev_io_set(&c->w, PQsocket(c->conn), EV_WRITE); break; case PGRES_POLLING_FAILED: lem_debug("PGRES_POLLING_FAILED"); goto error; case PGRES_POLLING_OK: lem_debug("PGRES_POLLING_OK"); return 1; #ifndef NDEBUG case PGRES_POLLING_ACTIVE: assert(0); #endif } c->T = T; ev_io_start(EV_G_ &c->w); lua_settop(T, 1); return lua_yield(T, 1); error: push_error(T, c->conn); return 2; } static void push_tuples(lua_State *T, PGresult *res) { int rows = PQntuples(res); int columns = PQnfields(res); int i; lua_createtable(T, rows, 0); for (i = 0; i < rows; i++) { int j; lua_createtable(T, columns, 0); for (j = 0; j < columns; j++) { if (PQgetisnull(res, i, j)) lua_pushnil(T); else lua_pushlstring(T, PQgetvalue(res, i, j), PQgetlength(res, i, j)); lua_rawseti(T, -2, j+1); } lua_rawseti(T, -2, i+1); } /* insert column names as "row 0" */ lua_createtable(T, columns, 0); for (i = 0; i < columns; i++) { lua_pushstring(T, PQfname(res, i)); lua_rawseti(T, -2, i+1); } lua_rawseti(T, -2, 0); } static void error_handler(EV_P_ struct ev_io *w, int revents) { struct connection *c = (struct connection *)w; (void)revents; if (PQconsumeInput(c->conn) == 0) { ev_io_stop(EV_A_ &c->w); lua_settop(c->T, 0); push_error(c->T, c->conn); lem_queue(c->T, 2); c->T = NULL; return; } while (!PQisBusy(c->conn)) { PGresult *res = PQgetResult(c->conn); if (res == NULL) { ev_io_stop(EV_A_ &c->w); lem_queue(c->T, 2); c->T = NULL; return; } PQclear(res); } } static void exec_handler(EV_P_ struct ev_io *w, int revents) { struct connection *c = (struct connection *)w; PGresult *res; (void)revents; if (PQconsumeInput(c->conn) != 1) { ev_io_stop(EV_A_ &c->w); lua_settop(c->T, 0); push_error(c->T, c->conn); lem_queue(c->T, 2); c->T = NULL; return; } while (!PQisBusy(c->conn)) { res = PQgetResult(c->conn); if (res == NULL) { ev_io_stop(EV_A_ &c->w); lem_debug("returning %d values", lua_gettop(c->T) - 1); lem_queue(c->T, lua_gettop(c->T) - 1); c->T = NULL; return; } switch (PQresultStatus(res)) { case PGRES_EMPTY_QUERY: lem_debug("PGRES_EMPTY_QUERY"); lua_settop(c->T, 0); lua_pushnil(c->T); lua_pushliteral(c->T, "empty query"); goto error; case PGRES_COMMAND_OK: lem_debug("PGRES_COMMAND_OK"); lua_pushboolean(c->T, 1); break; case PGRES_TUPLES_OK: lem_debug("PGRES_TUPLES_OK"); push_tuples(c->T, res); break; case PGRES_COPY_IN: lem_debug("PGRES_COPY_IN"); (void)PQsetnonblocking(c->conn, 1); case PGRES_COPY_OUT: lem_debug("PGRES_COPY_OUT"); PQclear(res); lua_pushboolean(c->T, 1); lem_queue(c->T, lua_gettop(c->T) - 1); c->T = NULL; return; case PGRES_BAD_RESPONSE: lem_debug("PGRES_BAD_RESPONSE"); lua_settop(c->T, 0); push_error(c->T, c->conn); goto error; case PGRES_NONFATAL_ERROR: lem_debug("PGRES_NONFATAL_ERROR"); break; case PGRES_FATAL_ERROR: lem_debug("PGRES_FATAL_ERROR"); lua_settop(c->T, 0); push_error(c->T, c->conn); goto error; default: lem_debug("unknown result status"); lua_settop(c->T, 0); lua_pushnil(c->T); lua_pushliteral(c->T, "unknown result status"); goto error; } PQclear(res); } lem_debug("busy"); return; error: PQclear(res); c->w.cb = error_handler; while (!PQisBusy(c->conn)) { res = PQgetResult(c->conn); if (res == NULL) { ev_io_stop(EV_A_ &c->w); lem_queue(c->T, 2); c->T = NULL; return; } PQclear(res); } } static int connection_exec(lua_State *T) { struct connection *c; const char *command; int n; luaL_checktype(T, 1, LUA_TUSERDATA); command = luaL_checkstring(T, 2); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } n = lua_gettop(T) - 2; if (n > 0) { const char **values = lem_xmalloc(n * sizeof(char *)); int *lengths = lem_xmalloc(n * sizeof(int)); int i; for (i = 0; i < n; i++) { size_t len; values[i] = lua_tolstring(T, i+3, &len); lengths[i] = len; } n = PQsendQueryParams(c->conn, command, n, NULL, values, lengths, NULL, 0); free(values); free(lengths); } else n = PQsendQuery(c->conn, command); if (n != 1) { lem_debug("PQsendQuery failed"); push_error(T, c->conn); return 2; } c->T = T; c->w.cb = exec_handler; c->w.events = EV_READ; ev_io_start(EV_G_ &c->w); lua_settop(T, 1); return lua_yield(T, 1); } static int connection_prepare(lua_State *T) { struct connection *c; const char *name; const char *query; luaL_checktype(T, 1, LUA_TUSERDATA); name = luaL_checkstring(T, 2); query = luaL_checkstring(T, 3); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } if (PQsendPrepare(c->conn, name, query, 0, NULL) != 1) { push_error(T, c->conn); return 2; } c->T = T; c->w.cb = exec_handler; c->w.events = EV_READ; ev_io_start(EV_G_ &c->w); lua_settop(T, 1); return lua_yield(T, 1); } static int connection_run(lua_State *T) { struct connection *c; const char *name; const char **values; int *lengths; int n; int i; luaL_checktype(T, 1, LUA_TUSERDATA); name = luaL_checkstring(T, 2); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } n = lua_gettop(T) - 2; values = lem_xmalloc(n * sizeof(char *)); lengths = lem_xmalloc(n * sizeof(int)); for (i = 0; i < n; i++) { size_t len; values[i] = lua_tolstring(T, i+3, &len); lengths[i] = len; } n = PQsendQueryPrepared(c->conn, name, n, values, lengths, NULL, 0); free(values); free(lengths); if (n != 1) { push_error(T, c->conn); return 2; } c->T = T; c->w.cb = exec_handler; c->w.events = EV_READ; ev_io_start(EV_G_ &c->w); lua_settop(T, 1); return lua_yield(T, 1); } static void put_handler(EV_P_ struct ev_io *w, int revents) { struct connection *c = (struct connection *)w; size_t len; const char *data; (void)revents; data = lua_tolstring(c->T, 2, &len); switch (PQputCopyData(c->conn, data, (int)len)) { case 1: /* data sent */ lem_debug("data sent"); ev_io_stop(EV_A_ &c->w); lua_settop(c->T, 0); lua_pushboolean(c->T, 1); lem_queue(c->T, 1); c->T = NULL; break; case 0: /* would block */ lem_debug("would block"); break; default: /* should be -1 for error */ lua_settop(c->T, 0); push_error(c->T, c->conn); lem_queue(c->T, 2); c->T = NULL; break; } } static int connection_put(lua_State *T) { struct connection *c; size_t len; const char *data; luaL_checktype(T, 1, LUA_TUSERDATA); data = luaL_checklstring(T, 2, &len); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } switch (PQputCopyData(c->conn, data, (int)len)) { case 1: /* data sent */ lem_debug("data sent"); lua_pushboolean(T, 1); return 1; case 0: /* would block */ lem_debug("would block"); break; default: /* should be -1 for error */ push_error(T, c->conn); return 2; } c->T = T; c->w.cb = put_handler; c->w.events = EV_WRITE; ev_io_start(EV_G_ &c->w); lua_settop(T, 2); return lua_yield(T, 2); } static void done_handler(EV_P_ struct ev_io *w, int revents) { struct connection *c = (struct connection *)w; const char *error = lua_tostring(c->T, 2); (void)revents; switch (PQputCopyEnd(c->conn, error)) { case 1: /* data sent */ lem_debug("data sent"); ev_io_stop(EV_A_ &c->w); lua_settop(c->T, 1); (void)PQsetnonblocking(c->conn, 0); c->w.cb = exec_handler; c->w.events = EV_READ; ev_io_start(EV_A_ &c->w); break; case 0: /* would block */ lem_debug("would block"); break; default: /* should be -1 for error */ ev_io_stop(EV_A_ &c->w); lua_settop(c->T, 0); push_error(c->T, c->conn); lem_queue(c->T, 2); c->T = NULL; break; } } static int connection_done(lua_State *T) { struct connection *c; const char *error; luaL_checktype(T, 1, LUA_TUSERDATA); error = luaL_optstring(T, 2, NULL); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } switch (PQputCopyEnd(c->conn, error)) { case 1: /* data sent */ lem_debug("data sent"); (void)PQsetnonblocking(c->conn, 0); c->T = T; c->w.cb = exec_handler; c->w.events = EV_READ; ev_io_start(EV_G_ &c->w); lua_settop(c->T, 1); return lua_yield(c->T, 1); case 0: /* would block */ lem_debug("would block"); break; default: /* should be -1 for error */ push_error(T, c->conn); return 2; } c->T = T; c->w.cb = done_handler; c->w.events = EV_WRITE; ev_io_start(EV_G_ &c->w); if (error == NULL) { lua_settop(T, 1); lua_pushnil(T); } else lua_settop(T, 2); return lua_yield(T, 2); } static void get_handler(EV_P_ struct ev_io *w, int revents) { struct connection *c = (struct connection *)w; char *buf; int ret; ret = PQgetCopyData(c->conn, &buf, 1); if (ret > 0) { lem_debug("got data"); ev_io_stop(EV_A_ &c->w); lua_pushlstring(c->T, buf, ret); PQfreemem(buf); lem_queue(c->T, 1); c->T = NULL; return; } switch (ret) { case 0: /* would block */ lem_debug("would block"); break; case -1: /* no more data */ lem_debug("no more data"); c->w.cb = exec_handler; exec_handler(EV_A_ &c->w, revents); break; default: /* should be -2 for error */ ev_io_stop(EV_A_ &c->w); lua_settop(c->T, 0); push_error(c->T, c->conn); lem_queue(c->T, 2); c->T = NULL; break; } } static int connection_get(lua_State *T) { struct connection *c; char *buf; int ret; luaL_checktype(T, 1, LUA_TUSERDATA); c = lua_touserdata(T, 1); if (c->T != NULL) { lua_pushnil(T); lua_pushliteral(T, "busy"); return 2; } if (c->conn == NULL) { lua_pushnil(T); lua_pushliteral(T, "closed"); return 2; } ret = PQgetCopyData(c->conn, &buf, 1); if (ret > 0) { lem_debug("got data"); lua_pushlstring(T, buf, ret); PQfreemem(buf); return 1; } switch (ret) { case 0: /* would block */ lem_debug("would block"); c->T = T; c->w.cb = get_handler; c->w.events = EV_READ; ev_io_start(EV_G_ &c->w); lua_settop(T, 1); return lua_yield(T, 1); case -1: /* no more data */ lem_debug("no more data"); break; default: /* should be -2 for error */ push_error(T, c->conn); return 2; } c->T = T; c->w.cb = exec_handler; c->w.events = EV_READ; ev_io_start(EV_G_ &c->w); /* TODO: it is necessary but kinda ugly to call * the exec_handler directly from here. * find a better solution... */ lua_settop(T, 1); exec_handler(EV_G_ &c->w, 0); return lua_yield(T, 1); } int luaopen_lem_postgres(lua_State *L) { lua_newtable(L); /* create Connection metatable mt */ lua_newtable(L); lua_pushvalue(L, -1); lua_setfield(L, -2, "__index"); /* mt.__gc = */ lua_pushcfunction(L, connection_gc); lua_setfield(L, -2, "__gc"); /* mt.close = */ lua_pushcfunction(L, connection_close); lua_setfield(L, -2, "close"); /* mt.connect = */ lua_pushvalue(L, -1); /* upvalue 1: Connection */ lua_pushcclosure(L, connection_connect, 1); lua_setfield(L, -3, "connect"); /* mt.reset = */ lua_pushcfunction(L, connection_reset); lua_setfield(L, -2, "reset"); /* mt.exec = */ lua_pushcfunction(L, connection_exec); lua_setfield(L, -2, "exec"); /* mt.prepare = */ lua_pushcfunction(L, connection_prepare); lua_setfield(L, -2, "prepare"); /* mt.run = */ lua_pushcfunction(L, connection_run); lua_setfield(L, -2, "run"); /* mt.put = */ lua_pushcfunction(L, connection_put); lua_setfield(L, -2, "put"); /* mt.done = */ lua_pushcfunction(L, connection_done); lua_setfield(L, -2, "done"); /* mt.get = */ lua_pushcfunction(L, connection_get); lua_setfield(L, -2, "get"); /* set Connection */ lua_setfield(L, -2, "Connection"); return 1; }