[irc/quakenet/newserv.git] / pqsql / pqsql.c

/* 
 * PQSQL module
 *
 * 99% of the handling is stolen from Q9.
 */

#include "../core/config.h"
#include "../core/error.h"
#include "../irc/irc_config.h"
#include "../core/events.h"
#include "../core/hooks.h"
#include "../core/nsmalloc.h"
#include "../lib/irc_string.h"
#include "../lib/version.h"
#include "../lib/strlfunc.h"
#include "pqsql.h"

#define BUILDING_DBAPI
#include "../dbapi/dbapi.h"

#include <stdlib.h>
#include <sys/poll.h>
#include <stdarg.h>
#include <string.h>

MODULE_VERSION("");

/* It's possible that we might want to do a very long query, longer than the
 * IRC-oriented SSTRING_MAX value.  One option would be to increase
 * SSTRING_MAX, but the whole purpose of sstring's is to efficiently deal
 * with situations where the malloc() padding overhead is large compared to
 * string length and strings are frequently recycled.  Since neither of
 * these are necessarily true for longer strings it makes more sense to use
 * malloc() for them. 
 *
 * So, query always points at the query string.  If it fitted in a sstring,
 * query_ss will point at the sstring for freeing purposes.  If query_ss is
 * NULL then it was malloc'd so should be free()'d directly.
 */
typedef struct pqasyncquery_s {
  sstring *query_ss;
  char *query;
  void *tag;
  PQQueryHandler handler;
  int flags;
  PQModuleIdentifier identifier;
  struct pqasyncquery_s *next;
} pqasyncquery_s;

typedef struct pqtableloaderinfo_s
{
    sstring *tablename;
    PQQueryHandler init, data, fini;
} pqtableloaderinfo_s;

pqasyncquery_s *queryhead = NULL, *querytail = NULL;

static int dbconnected = 0;
static PQModuleIdentifier moduleid = 0;
static PGconn *dbconn;

void dbhandler(int fd, short revents);
void pqstartloadtable(PGconn *dbconn, void *arg);
void dbstatus(int hooknum, void *arg);
void disconnectdb(void);
void connectdb(void);
char* pqlasterror(PGconn * pgconn);

void _init(void) {
  connectdb();
}

void _fini(void) {
  disconnectdb();

  nscheckfreeall(POOL_PQSQL);
}

PQModuleIdentifier pqgetid(void) {
  moduleid++;
  if(moduleid < 10)
    moduleid = 10;

  return moduleid;
}

void pqfreeid(PQModuleIdentifier identifier) {
  pqasyncquery_s *q, *p;
  
  if(identifier == 0 || !queryhead)
    return;

  if(queryhead->identifier == identifier) {
    (queryhead->handler)(NULL, queryhead->tag);
    queryhead->identifier = QH_ALREADYFIRED;
  }

  for(p=queryhead,q=queryhead->next;q;) {
    if(q->identifier == identifier) {
      (q->handler)(NULL, q->tag);
      p->next = q->next;

      if (q->query_ss) {
        freesstring(q->query_ss);
      } else {
        nsfree(POOL_PQSQL, q->query);
      }
      nsfree(POOL_PQSQL, q);
      q = p->next;
    } else {
      p = q;
      q = q->next;
    }
  }

  querytail = p;
}

void connectdb(void) {
  sstring *dbhost, *dbusername, *dbpassword, *dbdatabase, *dbport;
  char connectstr[1024];

  if(pqconnected())
    return;

  /* stolen from chanserv as I'm lazy */
  dbhost = getcopyconfigitem("pqsql", "host", "UNIX", HOSTLEN);
  dbusername = getcopyconfigitem("pqsql", "username", "newserv", 20);
  dbpassword = getcopyconfigitem("pqsql", "password", "moo", 20);
  dbdatabase = getcopyconfigitem("pqsql", "database", "newserv", 20);
  dbport = getcopyconfigitem("pqsql", "port", "431", 8);

  if(!dbhost || !dbusername || !dbpassword || !dbdatabase || !dbport) {
    /* freesstring allows NULL */
    freesstring(dbhost);
    freesstring(dbusername);
    freesstring(dbpassword);
    freesstring(dbdatabase);
    freesstring(dbport);
    return;
  }
  
  if (!strcmp(dbhost->content,"UNIX")) {
    snprintf(connectstr, sizeof(connectstr), "dbname=%s user=%s password=%s", dbdatabase->content, dbusername->content, dbpassword->content);
  } else {
    snprintf(connectstr, sizeof(connectstr), "host=%s port=%s dbname=%s user=%s password=%s", dbhost->content, dbport->content, dbdatabase->content, dbusername->content, dbpassword->content);
  }  

  freesstring(dbhost);
  freesstring(dbusername);
  freesstring(dbpassword);
  freesstring(dbdatabase);
  freesstring(dbport);

  Error("pqsql", ERR_INFO, "Attempting database connection: %s", connectstr);

  /* Blocking connect for now.. */
  dbconn = PQconnectdb(connectstr);
  
  if (!dbconn || (PQstatus(dbconn) != CONNECTION_OK)) {
    Error("pqsql", ERR_ERROR, "Unable to connect to db: %s", pqlasterror(dbconn));
    return;
  }
  Error("pqsql", ERR_INFO, "Connected!");

  dbconnected = 1;

  PQsetnonblocking(dbconn, 1);

  /* this kicks ass, thanks splidge! */
  registerhandler(PQsocket(dbconn), POLLIN, dbhandler);
  registerhook(HOOK_CORE_STATSREQUEST, dbstatus);
}

void dbhandler(int fd, short revents) {
  PGresult *res;
  pqasyncquery_s *qqp;

  if(revents & POLLIN) {
    PQconsumeInput(dbconn);
    
    if(!PQisBusy(dbconn)) { /* query is complete */
      if(queryhead->handler && queryhead->identifier != QH_ALREADYFIRED)
        (queryhead->handler)(dbconn, queryhead->tag);

      while((res = PQgetResult(dbconn))) {
        if(queryhead->identifier != QH_ALREADYFIRED) {
          switch(PQresultStatus(res)) {
            case PGRES_TUPLES_OK:
              Error("pqsql", ERR_WARNING, "Unhandled tuples output (query: %s)", queryhead->query);
              break;

            case PGRES_NONFATAL_ERROR:
            case PGRES_FATAL_ERROR:
              /* if a create query returns an error assume it went ok, paul will winge about this */
              if(!(queryhead->flags & DB_CREATE))
                Error("pqsql", ERR_WARNING, "Unhandled error response (query: %s)", queryhead->query);
              break;
	  
            default:
              break;
          }
        }
	
        PQclear(res);
      }

      /* Free the query and advance */
      qqp = queryhead;
      if(queryhead == querytail)
        querytail = NULL;

      queryhead = queryhead->next;

      if (qqp->query_ss) {
        freesstring(qqp->query_ss);
        qqp->query_ss=NULL;
        qqp->query=NULL;
      } else if (qqp->query) {
        nsfree(POOL_PQSQL, qqp->query);
        qqp->query=NULL;
      }
      nsfree(POOL_PQSQL, qqp);

      if(queryhead) { /* Submit the next query */	      
        PQsendQuery(dbconn, queryhead->query);
        PQflush(dbconn);
      }
    }
  }
}

/* sorry Q9 */
void pqasyncqueryf(int identifier, PQQueryHandler handler, void *tag, int flags, char *format, ...) {
  char querybuf[8192];
  va_list va;
  int len;
  pqasyncquery_s *qp;

  if(!pqconnected())
    return;

  va_start(va, format);
  len = vsnprintf(querybuf, sizeof(querybuf), format, va);
  va_end(va);

  /* PPA: no check here... */
  qp = (pqasyncquery_s *)nsmalloc(POOL_PQSQL, sizeof(pqasyncquery_s));

  if(!qp)
    Error("pqsql",ERR_STOP,"malloc() failed in pqsql.c");

  /* Use sstring or allocate (see above rant) */
  if (len > SSTRING_MAX) {
    qp->query = (char *)nsmalloc(POOL_PQSQL, len+1);
    strcpy(qp->query,querybuf);
    qp->query_ss=NULL;
  } else {
    qp->query_ss = getsstring(querybuf, len);
    qp->query = qp->query_ss->content;
  }
  qp->tag = tag;
  qp->handler = handler;
  qp->next = NULL; /* shove them at the end */
  qp->flags = flags;
  qp->identifier = identifier;

  if(querytail) {
    querytail->next = qp;
    querytail = qp;
  } else {
    querytail = queryhead = qp;
    PQsendQuery(dbconn, qp->query);
    PQflush(dbconn);
  }
}

void pqloadtable(char *tablename, PQQueryHandler init, PQQueryHandler data, PQQueryHandler fini)
{
  pqtableloaderinfo_s *tli;

  tli=(pqtableloaderinfo_s *)nsmalloc(POOL_PQSQL, sizeof(pqtableloaderinfo_s));
  tli->tablename=getsstring(tablename, 100);
  tli->init=init;
  tli->data=data;
  tli->fini=fini;
  pqasyncquery(pqstartloadtable, tli, "SELECT COUNT(*) FROM %s", tli->tablename->content);
}

void pqstartloadtable(PGconn *dbconn, void *arg)
{
  PGresult *res;
  unsigned long i, count, tablecrc;
  pqtableloaderinfo_s *tli = arg;

  res = PQgetResult(dbconn);

  if (PQresultStatus(res) != PGRES_TUPLES_OK && PQresultStatus(res) != PGRES_COMMAND_OK) {
    Error("pqsql", ERR_ERROR, "Error getting row count for %s.", tli->tablename->content);
    return;
  }

  if (PQnfields(res) != 1) {
    Error("pqsql", ERR_ERROR, "Count query format error for %s.", tli->tablename->content);
    return;
  }

  tablecrc=crc32(tli->tablename->content);
  count=strtoul(PQgetvalue(res, 0, 0), NULL, 10);
  PQclear(res);

  Error("pqsql", ERR_INFO, "Found %lu entries in table %s, scheduling load.", count, tli->tablename->content);

  pqasyncquery(tli->init, NULL, "BEGIN");
  pqasyncquery(NULL, NULL, "DECLARE table%lx%lx CURSOR FOR SELECT * FROM %s", tablecrc, count, tli->tablename->content);

  for (i=0;(count - i) > 1000; i+=1000)
    pqasyncquery(tli->data, NULL, "FETCH 1000 FROM table%lx%lx", tablecrc, count);

  pqasyncquery(tli->data, NULL, "FETCH ALL FROM table%lx%lx", tablecrc, count);

  pqasyncquery(NULL, NULL, "CLOSE table%lx%lx", tablecrc, count);
  pqasyncquery(tli->fini, NULL, "COMMIT");

  freesstring(tli->tablename);
  nsfree(POOL_PQSQL, tli);
}

void disconnectdb(void) {
  pqasyncquery_s *qqp = queryhead, *nqqp;

  if(!pqconnected())
    return;

  /* do this first else we may get conflicts */
  deregisterhandler(PQsocket(dbconn), 0);

  /* Throw all the queued queries away, beware of data malloc()ed inside the query item.. */
  while(qqp) {
    nqqp = qqp->next;
    if (qqp->query_ss) {
      freesstring(qqp->query_ss);
      qqp->query_ss=NULL;
      qqp->query=NULL;
    } else if (qqp->query) {
      nsfree(POOL_PQSQL, qqp->query);
      qqp->query=NULL;
    }
    nsfree(POOL_PQSQL, qqp);
    qqp = nqqp;
  }

  deregisterhook(HOOK_CORE_STATSREQUEST, dbstatus);
  PQfinish(dbconn);
  dbconn = NULL; /* hmm? */

  dbconnected = 0;
}

/* more stolen code from Q9 */
void dbstatus(int hooknum, void *arg) {
  if ((long)arg > 10) {
    int i = 0;
    pqasyncquery_s *qqp;
    char message[100];

    if(queryhead)
      for(qqp=queryhead;qqp;qqp=qqp->next)
        i++;
     
    snprintf(message, sizeof(message), "PQSQL   : %6d queries queued.",i);
    
    triggerhook(HOOK_CORE_STATSREPLY, message);
  }  
}

int pqconnected(void) {
  return dbconnected;
}

char* pqlasterror(PGconn * pgconn) {
  static char errormsg[PQ_ERRORMSG_LENGTH + 1];
  int i;
  if(!pgconn) 
    return "PGCONN NULL";
  strlcpy(errormsg, PQerrorMessage(pgconn), PQ_ERRORMSG_LENGTH);
  for(i=0;i<errormsg[i];i++) {
    if((errormsg[i] == '\r') || (errormsg[i] == '\n'))
      errormsg[i] = ' ';
    
  }
  return errormsg;
}

PQResult *pqgetresult(PGconn *c) {
  PQResult *r;
  if(!c)
    return NULL;

  r = (PQResult *)nsmalloc(POOL_PQSQL, sizeof(PQResult));
  r->row = -1;
  r->result = PQgetResult(c);
  r->rows = PQntuples(r->result);

  return r;
}

int pqfetchrow(PQResult *res) {
  if(res->row + 1 == res->rows)
    return 0;

  res->row++;

  return 1;
}

char *pqgetvalue(PQResult *res, int column) {
  return PQgetvalue(res->result, res->row, column);
}

void pqclear(PQResult *res) {
  if(!res)
    return;

  if(res->result)
    PQclear(res->result);

  nsfree(POOL_PQSQL, res);
}
Commit	Line	Data
e43481af CP	1	/*
	2	* PQSQL module
	3	*
	4	* 99% of the handling is stolen from Q9.
	5	*/
	6
	7	#include "../core/config.h"
	8	#include "../core/error.h"
	9	#include "../irc/irc_config.h"
	10	#include "../core/events.h"
	11	#include "../core/hooks.h"
76c8da69	12	#include "../core/nsmalloc.h"
c5373bb8	13	#include "../lib/irc_string.h"
87698d77	14	#include "../lib/version.h"
a321474b	15	#include "../lib/strlfunc.h"
e43481af CP	16	#include "pqsql.h"
e43481af CP	17
827cbcd7 CP	18	#define BUILDING_DBAPI
	19	#include "../dbapi/dbapi.h"
	20
e43481af CP	21	#include <stdlib.h>
	22	#include <sys/poll.h>
	23	#include <stdarg.h>
860fc324	24	#include <string.h>
e43481af	25
70b0a4e5	26	MODULE_VERSION("");
87698d77	27
59b7d3c2	28	/* It's possible that we might want to do a very long query, longer than the
	29	* IRC-oriented SSTRING_MAX value. One option would be to increase
	30	* SSTRING_MAX, but the whole purpose of sstring's is to efficiently deal
	31	* with situations where the malloc() padding overhead is large compared to
	32	* string length and strings are frequently recycled. Since neither of
	33	* these are necessarily true for longer strings it makes more sense to use
	34	* malloc() for them.
	35	*
	36	* So, query always points at the query string. If it fitted in a sstring,
	37	* query_ss will point at the sstring for freeing purposes. If query_ss is
	38	* NULL then it was malloc'd so should be free()'d directly.
	39	*/
e43481af	40	typedef struct pqasyncquery_s {
59b7d3c2	41	sstring *query_ss;
59b7d3c2	42	char *query;
e43481af CP	43	void *tag;
e43481af CP	44	PQQueryHandler handler;
67545367	45	int flags;
332870cd	46	PQModuleIdentifier identifier;
e43481af CP	47	struct pqasyncquery_s *next;
	48	} pqasyncquery_s;
	49
c5373bb8 C	50	typedef struct pqtableloaderinfo_s
	51	{
	52	sstring *tablename;
	53	PQQueryHandler init, data, fini;
	54	} pqtableloaderinfo_s;
	55
e43481af CP	56	pqasyncquery_s queryhead = NULL, querytail = NULL;
e43481af CP	57
ee8cd7d0	58	static int dbconnected = 0;
332870cd	59	static PQModuleIdentifier moduleid = 0;
ee8cd7d0	60	static PGconn *dbconn;
e43481af CP	61
e43481af CP	62	void dbhandler(int fd, short revents);
c5373bb8	63	void pqstartloadtable(PGconn dbconn, void arg);
e43481af CP	64	void dbstatus(int hooknum, void *arg);
	65	void disconnectdb(void);
	66	void connectdb(void);
ee8cd7d0	67	char* pqlasterror(PGconn * pgconn);
e43481af CP	68
	69	void _init(void) {
	70	connectdb();
	71	}
	72
	73	void _fini(void) {
	74	disconnectdb();
76c8da69 CP	75
76c8da69 CP	76	nscheckfreeall(POOL_PQSQL);
e43481af CP	77	}
e43481af CP	78
332870cd CP	79	PQModuleIdentifier pqgetid(void) {
	80	moduleid++;
	81	if(moduleid < 10)
	82	moduleid = 10;
	83
	84	return moduleid;
	85	}
	86
	87	void pqfreeid(PQModuleIdentifier identifier) {
	88	pqasyncquery_s q, p;
	89
	90	if(identifier == 0 \|\| !queryhead)
	91	return;
	92
	93	if(queryhead->identifier == identifier) {
	94	(queryhead->handler)(NULL, queryhead->tag);
	95	queryhead->identifier = QH_ALREADYFIRED;
	96	}
	97
	98	for(p=queryhead,q=queryhead->next;q;) {
	99	if(q->identifier == identifier) {
	100	(q->handler)(NULL, q->tag);
	101	p->next = q->next;
	102
	103	if (q->query_ss) {
	104	freesstring(q->query_ss);
	105	} else {
76c8da69	106	nsfree(POOL_PQSQL, q->query);
332870cd	107	}
76c8da69	108	nsfree(POOL_PQSQL, q);
332870cd CP	109	q = p->next;
	110	} else {
	111	p = q;
	112	q = q->next;
	113	}
	114	}
	115
	116	querytail = p;
	117	}
	118
e43481af CP	119	void connectdb(void) {
	120	sstring dbhost, dbusername, dbpassword, dbdatabase, *dbport;
	121	char connectstr[1024];
	122
	123	if(pqconnected())
	124	return;
	125
	126	/* stolen from chanserv as I'm lazy */
a321474b	127	dbhost = getcopyconfigitem("pqsql", "host", "UNIX", HOSTLEN);
e43481af CP	128	dbusername = getcopyconfigitem("pqsql", "username", "newserv", 20);
	129	dbpassword = getcopyconfigitem("pqsql", "password", "moo", 20);
	130	dbdatabase = getcopyconfigitem("pqsql", "database", "newserv", 20);
	131	dbport = getcopyconfigitem("pqsql", "port", "431", 8);
	132
	133	if(!dbhost \|\| !dbusername \|\| !dbpassword \|\| !dbdatabase \|\| !dbport) {
	134	/* freesstring allows NULL */
	135	freesstring(dbhost);
	136	freesstring(dbusername);
	137	freesstring(dbpassword);
	138	freesstring(dbdatabase);
	139	freesstring(dbport);
	140	return;
	141	}
e43481af	142
a321474b	143	if (!strcmp(dbhost->content,"UNIX")) {
	144	snprintf(connectstr, sizeof(connectstr), "dbname=%s user=%s password=%s", dbdatabase->content, dbusername->content, dbpassword->content);
	145	} else {
	146	snprintf(connectstr, sizeof(connectstr), "host=%s port=%s dbname=%s user=%s password=%s", dbhost->content, dbport->content, dbdatabase->content, dbusername->content, dbpassword->content);
	147	}
	148
e43481af CP	149	freesstring(dbhost);
	150	freesstring(dbusername);
	151	freesstring(dbpassword);
	152	freesstring(dbdatabase);
	153	freesstring(dbport);
	154
	155	Error("pqsql", ERR_INFO, "Attempting database connection: %s", connectstr);
	156
	157	/* Blocking connect for now.. */
	158	dbconn = PQconnectdb(connectstr);
	159
7cc1a026	160	if (!dbconn \|\| (PQstatus(dbconn) != CONNECTION_OK)) {
47d8702b	161	Error("pqsql", ERR_ERROR, "Unable to connect to db: %s", pqlasterror(dbconn));
e43481af	162	return;
7cc1a026	163	}
4f981d28 P	164	Error("pqsql", ERR_INFO, "Connected!");
4f981d28 P	165
e43481af CP	166	dbconnected = 1;
	167
	168	PQsetnonblocking(dbconn, 1);
	169
	170	/* this kicks ass, thanks splidge! */
	171	registerhandler(PQsocket(dbconn), POLLIN, dbhandler);
	172	registerhook(HOOK_CORE_STATSREQUEST, dbstatus);
	173	}
	174
	175	void dbhandler(int fd, short revents) {
	176	PGresult *res;
	177	pqasyncquery_s *qqp;
	178
	179	if(revents & POLLIN) {
	180	PQconsumeInput(dbconn);
	181
	182	if(!PQisBusy(dbconn)) { /* query is complete */
332870cd	183	if(queryhead->handler && queryhead->identifier != QH_ALREADYFIRED)
e43481af CP	184	(queryhead->handler)(dbconn, queryhead->tag);
	185
	186	while((res = PQgetResult(dbconn))) {
332870cd CP	187	if(queryhead->identifier != QH_ALREADYFIRED) {
	188	switch(PQresultStatus(res)) {
	189	case PGRES_TUPLES_OK:
	190	Error("pqsql", ERR_WARNING, "Unhandled tuples output (query: %s)", queryhead->query);
	191	break;
	192
	193	case PGRES_NONFATAL_ERROR:
	194	case PGRES_FATAL_ERROR:
	195	/* if a create query returns an error assume it went ok, paul will winge about this */
ee8cd7d0	196	if(!(queryhead->flags & DB_CREATE))
332870cd CP	197	Error("pqsql", ERR_WARNING, "Unhandled error response (query: %s)", queryhead->query);
332870cd CP	198	break;
e43481af	199
332870cd CP	200	default:
	201	break;
	202	}
e43481af CP	203	}
	204
	205	PQclear(res);
	206	}
	207
	208	/* Free the query and advance */
	209	qqp = queryhead;
	210	if(queryhead == querytail)
	211	querytail = NULL;
	212
	213	queryhead = queryhead->next;
	214
59b7d3c2	215	if (qqp->query_ss) {
	216	freesstring(qqp->query_ss);
	217	qqp->query_ss=NULL;
	218	qqp->query=NULL;
	219	} else if (qqp->query) {
76c8da69	220	nsfree(POOL_PQSQL, qqp->query);
59b7d3c2	221	qqp->query=NULL;
59b7d3c2	222	}
76c8da69	223	nsfree(POOL_PQSQL, qqp);
e43481af CP	224
e43481af CP	225	if(queryhead) { /* Submit the next query */
860fc324	226	PQsendQuery(dbconn, queryhead->query);
e43481af CP	227	PQflush(dbconn);
	228	}
	229	}
	230	}
	231	}
	232
	233	/* sorry Q9 */
332870cd	234	void pqasyncqueryf(int identifier, PQQueryHandler handler, void tag, int flags, char format, ...) {
e43481af CP	235	char querybuf[8192];
	236	va_list va;
	237	int len;
	238	pqasyncquery_s *qp;
	239
	240	if(!pqconnected())
	241	return;
	242
	243	va_start(va, format);
	244	len = vsnprintf(querybuf, sizeof(querybuf), format, va);
	245	va_end(va);
	246
	247	/* PPA: no check here... */
76c8da69	248	qp = (pqasyncquery_s *)nsmalloc(POOL_PQSQL, sizeof(pqasyncquery_s));
59b7d3c2	249
67545367	250	if(!qp)
59b7d3c2	251	Error("pqsql",ERR_STOP,"malloc() failed in pqsql.c");
67545367	252
59b7d3c2	253	/* Use sstring or allocate (see above rant) */
59b7d3c2	254	if (len > SSTRING_MAX) {
76c8da69	255	qp->query = (char *)nsmalloc(POOL_PQSQL, len+1);
59b7d3c2	256	strcpy(qp->query,querybuf);
	257	qp->query_ss=NULL;
	258	} else {
	259	qp->query_ss = getsstring(querybuf, len);
	260	qp->query = qp->query_ss->content;
	261	}
e43481af CP	262	qp->tag = tag;
	263	qp->handler = handler;
	264	qp->next = NULL; /* shove them at the end */
67545367	265	qp->flags = flags;
332870cd	266	qp->identifier = identifier;
e43481af CP	267
	268	if(querytail) {
	269	querytail->next = qp;
	270	querytail = qp;
	271	} else {
	272	querytail = queryhead = qp;
860fc324	273	PQsendQuery(dbconn, qp->query);
e43481af CP	274	PQflush(dbconn);
	275	}
	276	}
	277
c5373bb8 C	278	void pqloadtable(char *tablename, PQQueryHandler init, PQQueryHandler data, PQQueryHandler fini)
	279	{
	280	pqtableloaderinfo_s *tli;
	281
76c8da69	282	tli=(pqtableloaderinfo_s *)nsmalloc(POOL_PQSQL, sizeof(pqtableloaderinfo_s));
c5373bb8 C	283	tli->tablename=getsstring(tablename, 100);
	284	tli->init=init;
	285	tli->data=data;
	286	tli->fini=fini;
	287	pqasyncquery(pqstartloadtable, tli, "SELECT COUNT(*) FROM %s", tli->tablename->content);
	288	}
	289
	290	void pqstartloadtable(PGconn dbconn, void arg)
	291	{
	292	PGresult *res;
	293	unsigned long i, count, tablecrc;
	294	pqtableloaderinfo_s *tli = arg;
	295
	296	res = PQgetResult(dbconn);
	297
	298	if (PQresultStatus(res) != PGRES_TUPLES_OK && PQresultStatus(res) != PGRES_COMMAND_OK) {
	299	Error("pqsql", ERR_ERROR, "Error getting row count for %s.", tli->tablename->content);
	300	return;
	301	}
	302
	303	if (PQnfields(res) != 1) {
	304	Error("pqsql", ERR_ERROR, "Count query format error for %s.", tli->tablename->content);
	305	return;
	306	}
	307
	308	tablecrc=crc32(tli->tablename->content);
	309	count=strtoul(PQgetvalue(res, 0, 0), NULL, 10);
	310	PQclear(res);
	311
	312	Error("pqsql", ERR_INFO, "Found %lu entries in table %s, scheduling load.", count, tli->tablename->content);
	313
	314	pqasyncquery(tli->init, NULL, "BEGIN");
	315	pqasyncquery(NULL, NULL, "DECLARE table%lx%lx CURSOR FOR SELECT * FROM %s", tablecrc, count, tli->tablename->content);
	316
	317	for (i=0;(count - i) > 1000; i+=1000)
	318	pqasyncquery(tli->data, NULL, "FETCH 1000 FROM table%lx%lx", tablecrc, count);
	319
	320	pqasyncquery(tli->data, NULL, "FETCH ALL FROM table%lx%lx", tablecrc, count);
	321
	322	pqasyncquery(NULL, NULL, "CLOSE table%lx%lx", tablecrc, count);
	323	pqasyncquery(tli->fini, NULL, "COMMIT");
	324
	325	freesstring(tli->tablename);
76c8da69	326	nsfree(POOL_PQSQL, tli);
c5373bb8 C	327	}
c5373bb8 C	328
e43481af CP	329	void disconnectdb(void) {
	330	pqasyncquery_s qqp = queryhead, nqqp;
	331
	332	if(!pqconnected())
	333	return;
	334
	335	/* do this first else we may get conflicts */
	336	deregisterhandler(PQsocket(dbconn), 0);
	337
	338	/* Throw all the queued queries away, beware of data malloc()ed inside the query item.. */
	339	while(qqp) {
	340	nqqp = qqp->next;
59b7d3c2	341	if (qqp->query_ss) {
860fc324	342	freesstring(qqp->query_ss);
59b7d3c2	343	qqp->query_ss=NULL;
	344	qqp->query=NULL;
	345	} else if (qqp->query) {
76c8da69	346	nsfree(POOL_PQSQL, qqp->query);
59b7d3c2	347	qqp->query=NULL;
59b7d3c2	348	}
76c8da69	349	nsfree(POOL_PQSQL, qqp);
e43481af CP	350	qqp = nqqp;
	351	}
	352
	353	deregisterhook(HOOK_CORE_STATSREQUEST, dbstatus);
	354	PQfinish(dbconn);
	355	dbconn = NULL; /* hmm? */
	356
	357	dbconnected = 0;
	358	}
	359
	360	/* more stolen code from Q9 */
	361	void dbstatus(int hooknum, void *arg) {
c3db6f7e	362	if ((long)arg > 10) {
e43481af CP	363	int i = 0;
	364	pqasyncquery_s *qqp;
	365	char message[100];
	366
	367	if(queryhead)
	368	for(qqp=queryhead;qqp;qqp=qqp->next)
	369	i++;
	370
6b5b3c37	371	snprintf(message, sizeof(message), "PQSQL : %6d queries queued.",i);
e43481af CP	372
	373	triggerhook(HOOK_CORE_STATSREPLY, message);
	374	}
	375	}
	376
	377	int pqconnected(void) {
	378	return dbconnected;
	379	}
47d8702b	380
1e5a45f5 P	381	char* pqlasterror(PGconn * pgconn) {
1e5a45f5 P	382	static char errormsg[PQ_ERRORMSG_LENGTH + 1];
47d8702b	383	int i;
1e5a45f5 P	384	if(!pgconn)
	385	return "PGCONN NULL";
	386	strlcpy(errormsg, PQerrorMessage(pgconn), PQ_ERRORMSG_LENGTH);
	387	for(i=0;i<errormsg[i];i++) {
	388	if((errormsg[i] == '\r') \|\| (errormsg[i] == '\n'))
47d8702b	389	errormsg[i] = ' ';
1e5a45f5	390
47d8702b P	391	}
	392	return errormsg;
	393	}
ee8cd7d0 CP	394
	395	PQResult pqgetresult(PGconn c) {
	396	PQResult *r;
	397	if(!c)
	398	return NULL;
	399
76c8da69	400	r = (PQResult *)nsmalloc(POOL_PQSQL, sizeof(PQResult));
ee8cd7d0 CP	401	r->row = -1;
	402	r->result = PQgetResult(c);
	403	r->rows = PQntuples(r->result);
	404
	405	return r;
	406	}
	407
	408	int pqfetchrow(PQResult *res) {
	409	if(res->row + 1 == res->rows)
	410	return 0;
	411
	412	res->row++;
	413
	414	return 1;
	415	}
	416
	417	char pqgetvalue(PQResult res, int column) {
	418	return PQgetvalue(res->result, res->row, column);
	419	}
	420
	421	void pqclear(PQResult *res) {
	422	if(!res)
	423	return;
	424
	425	if(res->result)
	426	PQclear(res->result);
	427
76c8da69	428	nsfree(POOL_PQSQL, res);
ee8cd7d0	429	}