X-Git-Url: https://git.netwichtig.de/gitweb/?a=blobdiff_plain;f=src%2Fmodules%2Fextra%2Fm_pgsql.cpp;h=1e73c0143768655a81d799db9687561e625e01b4;hb=3ccf0065d43db80f31c6404aeac4d65551481508;hp=85d092241d2074627e42d00f0ae9beb60684724b;hpb=11e45f2cb78c0667e2c7c7e2370944bf64b140b8;p=user%2Fhenk%2Fcode%2Finspircd.git diff --git a/src/modules/extra/m_pgsql.cpp b/src/modules/extra/m_pgsql.cpp index 85d092241..1e73c0143 100644 --- a/src/modules/extra/m_pgsql.cpp +++ b/src/modules/extra/m_pgsql.cpp @@ -1,26 +1,34 @@ -/* +------------------------------------+ - * | Inspire Internet Relay Chat Daemon | - * +------------------------------------+ +/* + * InspIRCd -- Internet Relay Chat Daemon * - * InspIRCd: (C) 2002-2010 InspIRCd Development Team - * See: http://wiki.inspircd.org/Credits + * Copyright (C) 2009-2010 Daniel De Graaf + * Copyright (C) 2006-2007, 2009 Dennis Friis + * Copyright (C) 2006-2007, 2009 Craig Edwards + * Copyright (C) 2008 Robin Burchell + * Copyright (C) 2008 Thomas Stagner + * Copyright (C) 2006 Oliver Lupton * - * This program is free but copyrighted software; see - * the file COPYING for details. + * This file is part of InspIRCd. InspIRCd 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, version 2. * - * --------------------------------------------------- + * This program 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 this program. If not, see . */ + #include "inspircd.h" #include -#include #include -#include "sql.h" +#include "modules/sql.h" -/* $ModDesc: PostgreSQL Service Provider module for all other m_sql* modules, uses v2 of the SQL API */ /* $CompileFlags: -Iexec("pg_config --includedir") eval("my $s = `pg_config --version`;$s =~ /^.*?(\d+)\.(\d+)\.(\d+).*?$/;my $v = hex(sprintf("0x%02x%02x%02x", $1, $2, $3));print "-DPGSQL_HAS_ESCAPECONN" if(($v >= 0x080104) || ($v >= 0x07030F && $v < 0x070400) || ($v >= 0x07040D && $v < 0x080000) || ($v >= 0x080008 && $v < 0x080100));") */ /* $LinkerFlags: -Lexec("pg_config --libdir") -lpq */ -/* $ModDep: m_sqlv2.h */ /* SQLConn rewritten by peavey to * use EventHandler instead of @@ -33,7 +41,7 @@ class SQLConn; class ModulePgSQL; -typedef std::map ConnMap; +typedef insp::flat_map ConnMap; /* CREAD, Connecting and wants read event * CWRITE, Connecting and wants write event @@ -49,12 +57,18 @@ class ReconnectTimer : public Timer private: ModulePgSQL* mod; public: - ReconnectTimer(ModulePgSQL* m) : Timer(5, ServerInstance->Time(), false), mod(m) + ReconnectTimer(ModulePgSQL* m) : Timer(5, false), mod(m) { } - virtual void Tick(time_t TIME); + bool Tick(time_t TIME); }; +struct QueueItem +{ + SQLQuery* c; + std::string q; + QueueItem(SQLQuery* C, const std::string& Q) : c(C), q(Q) {} +}; /** PgSQLresult is a subclass of the mostly-pure-virtual class SQLresult. * All SQL providers must create their own subclass and define it's methods using that @@ -81,12 +95,12 @@ class PgSQLresult : public SQLResult PQclear(res); } - virtual int Rows() + int Rows() { return rows; } - virtual void GetCols(std::vector& result) + void GetCols(std::vector& result) { result.resize(PQnfields(res)); for(unsigned int i=0; i < result.size(); i++) @@ -95,7 +109,7 @@ class PgSQLresult : public SQLResult } } - virtual SQLEntry GetValue(int row, int column) + SQLEntry GetValue(int row, int column) { char* v = PQgetvalue(res, row, column); if (!v || PQgetisnull(res, row, column)) @@ -104,7 +118,7 @@ class PgSQLresult : public SQLResult return SQLEntry(std::string(v, PQgetlength(res, row, column))); } - virtual bool GetRow(SQLEntries& result) + bool GetRow(SQLEntries& result) { if (currentrow >= PQntuples(res)) return false; @@ -124,22 +138,19 @@ class PgSQLresult : public SQLResult */ class SQLConn : public SQLProvider, public EventHandler { - private: + public: reference conf; /* The entry */ - std::deque queue; + std::deque queue; PGconn* sql; /* PgSQL database connection handle */ SQLstatus status; /* PgSQL database connection status */ - SQLQuery* qinprog; /* If there is currently a query in progress */ - time_t idle; /* Time we last heard from the database */ + QueueItem qinprog; /* If there is currently a query in progress */ - public: SQLConn(Module* Creator, ConfigTag* tag) - : SQLProvider(Creator, "SQL/" + tag->getString("id")), conf(tag), sql(NULL), status(CWRITE), qinprog(NULL) + : SQLProvider(Creator, "SQL/" + tag->getString("id")), conf(tag), sql(NULL), status(CWRITE), qinprog(NULL, "") { - idle = ServerInstance->Time(); if (!DoConnect()) { - ServerInstance->Logs->Log("m_pgsql",DEFAULT, "WARNING: Could not connect to database " + tag->getString("id")); + ServerInstance->Logs->Log(MODNAME, LOG_DEFAULT, "WARNING: Could not connect to database " + tag->getString("id")); DelayReconnect(); } } @@ -154,20 +165,20 @@ class SQLConn : public SQLProvider, public EventHandler ~SQLConn() { SQLerror err(SQL_BAD_DBID); - if (qinprog) + if (qinprog.c) { - qinprog->OnError(err); - delete qinprog; + qinprog.c->OnError(err); + delete qinprog.c; } - for(std::deque::iterator i = queue.begin(); i != queue.end(); i++) + for(std::deque::iterator i = queue.begin(); i != queue.end(); i++) { - SQLQuery* q = *i; + SQLQuery* q = i->c; q->OnError(err); delete q; } } - virtual void HandleEvent(EventType et, int errornum) + void HandleEvent(EventType et, int errornum) { switch (et) { @@ -229,9 +240,9 @@ class SQLConn : public SQLProvider, public EventHandler if(this->fd <= -1) return false; - if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_WRITE | FD_WANT_NO_READ)) + if (!SocketEngine::AddFd(this, FD_WANT_NO_WRITE | FD_WANT_NO_READ)) { - ServerInstance->Logs->Log("m_pgsql",DEBUG, "BUG: Couldn't add pgsql socket to socket engine"); + ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Couldn't add pgsql socket to socket engine"); return false; } @@ -244,17 +255,17 @@ class SQLConn : public SQLProvider, public EventHandler switch(PQconnectPoll(sql)) { case PGRES_POLLING_WRITING: - ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ); + SocketEngine::ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ); status = CWRITE; return true; case PGRES_POLLING_READING: - ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + SocketEngine::ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); status = CREAD; return true; case PGRES_POLLING_FAILED: return false; case PGRES_POLLING_OK: - ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + SocketEngine::ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); status = WWRITE; DoConnectedPoll(); default: @@ -265,7 +276,7 @@ class SQLConn : public SQLProvider, public EventHandler void DoConnectedPoll() { restart: - while (!qinprog && !queue.empty()) + while (qinprog.q.empty() && !queue.empty()) { /* There's no query currently in progress, and there's queries in the queue. */ DoQuery(queue.front()); @@ -274,16 +285,11 @@ restart: if (PQconsumeInput(sql)) { - /* We just read stuff from the server, that counts as it being alive - * so update the idle-since time :p - */ - idle = ServerInstance->Time(); - if (PQisBusy(sql)) { /* Nothing happens here */ } - else if (qinprog) + else if (qinprog.c) { /* Fetch the result.. */ PGresult* result = PQgetResult(sql); @@ -309,18 +315,22 @@ restart: case PGRES_FATAL_ERROR: { SQLerror err(SQL_QREPLY_FAIL, PQresultErrorMessage(result)); - qinprog->OnError(err); + qinprog.c->OnError(err); break; } default: /* Other values are not errors */ - qinprog->OnResult(reply); + qinprog.c->OnResult(reply); } - delete qinprog; - qinprog = NULL; + delete qinprog.c; + qinprog = QueueItem(NULL, ""); goto restart; } + else + { + qinprog.q.clear(); + } } else { @@ -338,17 +348,17 @@ restart: switch(PQresetPoll(sql)) { case PGRES_POLLING_WRITING: - ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ); + SocketEngine::ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ); status = CWRITE; return DoPoll(); case PGRES_POLLING_READING: - ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + SocketEngine::ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); status = CREAD; return true; case PGRES_POLLING_FAILED: return false; case PGRES_POLLING_OK: - ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); + SocketEngine::ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE); status = WWRITE; DoConnectedPoll(); default: @@ -374,7 +384,20 @@ restart: } } - virtual std::string FormatQuery(const std::string& q, const ParamL& p) + void submit(SQLQuery *req, const std::string& q) + { + if (qinprog.q.empty()) + { + DoQuery(QueueItem(req,q)); + } + else + { + // wait your turn. + queue.push_back(QueueItem(req,q)); + } + } + + void submit(SQLQuery *req, const std::string& q, const ParamL& p) { std::string res; unsigned int param = 0; @@ -384,27 +407,26 @@ restart: res.push_back(q[i]); else { - // TODO numbered parameter support ('?1') if (param < p.size()) { std::string parm = p[param++]; - char buffer[MAXBUF]; + std::vector buffer(parm.length() * 2 + 1); #ifdef PGSQL_HAS_ESCAPECONN int error; - PQescapeStringConn(sql, buffer, parm.c_str(), parm.length(), &error); + size_t escapedsize = PQescapeStringConn(sql, &buffer[0], parm.data(), parm.length(), &error); if (error) - ServerInstance->Logs->Log("m_pgsql", DEBUG, "BUG: Apparently PQescapeStringConn() failed"); + ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Apparently PQescapeStringConn() failed"); #else - PQescapeString (buffer, parm.c_str(), parm.length()); + size_t escapedsize = PQescapeString(&buffer[0], parm.data(), parm.length()); #endif - res.append(buffer); + res.append(&buffer[0], escapedsize); } } } - return res; + submit(req, res); } - std::string FormatQuery(const std::string& q, const ParamM& p) + void submit(SQLQuery *req, const std::string& q, const ParamM& p) { std::string res; for(std::string::size_type i = 0; i < q.length(); i++) @@ -415,7 +437,7 @@ restart: { std::string field; i++; - while (i < q.length() && isalpha(q[i])) + while (i < q.length() && isalnum(q[i])) field.push_back(q[i++]); i--; @@ -423,61 +445,48 @@ restart: if (it != p.end()) { std::string parm = it->second; - char buffer[MAXBUF]; + std::vector buffer(parm.length() * 2 + 1); #ifdef PGSQL_HAS_ESCAPECONN int error; - PQescapeStringConn(sql, buffer, parm.c_str(), parm.length(), &error); + size_t escapedsize = PQescapeStringConn(sql, &buffer[0], parm.data(), parm.length(), &error); if (error) - ServerInstance->Logs->Log("m_pgsql", DEBUG, "BUG: Apparently PQescapeStringConn() failed"); + ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Apparently PQescapeStringConn() failed"); #else - PQescapeString (buffer, parm.c_str(), parm.length()); + size_t escapedsize = PQescapeString(&buffer[0], parm.data(), parm.length()); #endif - res.append(buffer); + res.append(&buffer[0], escapedsize); } } } - return res; + submit(req, res); } - virtual void submit(SQLQuery *req) - { - if (qinprog) - { - // wait your turn. - queue.push_back(req); - } - else - { - DoQuery(req); - } - } - - void DoQuery(SQLQuery* req) + void DoQuery(const QueueItem& req) { if (status != WREAD && status != WWRITE) { // whoops, not connected... SQLerror err(SQL_BAD_CONN); - req->OnError(err); - delete req; + req.c->OnError(err); + delete req.c; return; } - if(PQsendQuery(sql, req->query.c_str())) + if(PQsendQuery(sql, req.q.c_str())) { qinprog = req; } else { SQLerror err(SQL_QSEND_FAIL, PQerrorMessage(sql)); - req->OnError(err); - delete req; + req.c->OnError(err); + delete req.c; } } void Close() { - ServerInstance->SE->DelFd(this); + SocketEngine::DelFd(this); if(sql) { @@ -494,25 +503,17 @@ class ModulePgSQL : public Module ReconnectTimer* retimer; ModulePgSQL() + : retimer(NULL) { } - void init() - { - ReadConf(); - - Implementation eventlist[] = { I_OnUnloadModule, I_OnRehash }; - ServerInstance->Modules->Attach(eventlist, this, 2); - } - - virtual ~ModulePgSQL() + ~ModulePgSQL() { - if (retimer) - ServerInstance->Timers->DelTimer(retimer); + delete retimer; ClearAllConnections(); } - virtual void OnRehash(User* user) + void ReadConfig(ConfigStatus& status) CXX11_OVERRIDE { ReadConf(); } @@ -553,21 +554,46 @@ class ModulePgSQL : public Module connections.clear(); } - void OnUnloadModule(Module* mod) + void OnUnloadModule(Module* mod) CXX11_OVERRIDE { - // TODO cancel queries that will have a bad vtable + SQLerror err(SQL_BAD_DBID); + for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++) + { + SQLConn* conn = i->second; + if (conn->qinprog.c && conn->qinprog.c->creator == mod) + { + conn->qinprog.c->OnError(err); + delete conn->qinprog.c; + conn->qinprog.c = NULL; + } + std::deque::iterator j = conn->queue.begin(); + while (j != conn->queue.end()) + { + SQLQuery* q = j->c; + if (q->creator == mod) + { + q->OnError(err); + delete q; + j = conn->queue.erase(j); + } + else + j++; + } + } } - Version GetVersion() + Version GetVersion() CXX11_OVERRIDE { return Version("PostgreSQL Service Provider module for all other m_sql* modules, uses v2 of the SQL API", VF_VENDOR); } }; -void ReconnectTimer::Tick(time_t time) +bool ReconnectTimer::Tick(time_t time) { mod->retimer = NULL; mod->ReadConf(); + delete this; + return false; } void SQLConn::DelayReconnect() @@ -581,7 +607,7 @@ void SQLConn::DelayReconnect() if (!mod->retimer) { mod->retimer = new ReconnectTimer(mod); - ServerInstance->Timers->AddTimer(mod->retimer); + ServerInstance->Timers.AddTimer(mod->retimer); } } }