citadel/modules/eventclient/serv_eventclient.c

   1 /*
   2  * Copyright (c) 1998-2009 by the citadel.org team
   3  *
   4  *  This program is free software; you can redistribute it and/or modify
   5  *  it under the terms of the GNU General Public License as published by
   6  *  the Free Software Foundation; either version 3 of the License, or
   7  *  (at your option) any later version.
   8  *
   9  *  This program is distributed in the hope that it will be useful,
  10  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  *  GNU General Public License for more details.
  13  *
  14  *  You should have received a copy of the GNU General Public License
  15  *  along with this program; if not, write to the Free Software
  16  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  17  */
  18
  19 #include "sysdep.h"
  20 #include <stdlib.h>
  21 #include <unistd.h>
  22 #include <stdio.h>
  23 #include <termios.h>
  24 #include <fcntl.h>
  25 #include <signal.h>
  26 #include <pwd.h>
  27 #include <errno.h>
  28 #include <sys/types.h>
  29 #include <syslog.h>
  30
  31 #if TIME_WITH_SYS_TIME
  32 # include <sys/time.h>
  33 # include <time.h>
  34 #else
  35 # if HAVE_SYS_TIME_H
  36 #  include <sys/time.h>
  37 # else
  38 #  include <time.h>
  39 # endif
  40 #endif
  41 #include <sys/wait.h>
  42 #include <ctype.h>
  43 #include <string.h>
  44 #include <limits.h>
  45 #include <sys/socket.h>
  46 #include <netinet/in.h>
  47 #include <assert.h>
  48 #include <arpa/inet.h>
  49 #include <libcitadel.h>
  50 #include <curl/curl.h>
  51 #include <curl/multi.h>
  52 #include "citadel.h"
  53 #include "server.h"
  54 #include "citserver.h"
  55 #include "support.h"
  56
  57 #include "ctdl_module.h"
  58
  59 #include "event_client.h"
  60 #include "serv_curl.h"
  61
  62 ev_loop *event_base;
  63
  64 long EvIDSource = 1;
  65 /*****************************************************************************
  66  *                   libevent / curl integration                             *
  67  *****************************************************************************/
  68 #define MOPT(s, v)                                                      \
  69         do {                                                            \
  70                 sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
  71                 if (sta) {                                              \
  72                         syslog(LOG_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \
  73                         exit (1);                                       \
  74                 }                                                       \
  75         } while (0)
  76
  77 typedef struct _evcurl_global_data {
  78         int magic;
  79         CURLM *mhnd;
  80         ev_timer timeev;
  81         int nrun;
  82 } evcurl_global_data;
  83
  84 ev_async WakeupCurl;
  85 evcurl_global_data global;
  86
  87 static void
  88 gotstatus(int nnrun)
  89 {
  90         CURLMsg *msg;
  91         int nmsg;
  92
  93         global.nrun = nnrun;
  94
  95         syslog(LOG_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
  96         while ((msg = curl_multi_info_read(global.mhnd, &nmsg))) {
  97                 syslog(LOG_ERR, "EVCURL: got curl multi_info message msg=%d\n", msg->msg);
  98                 if (CURLMSG_DONE == msg->msg) {
  99                         CURL *chnd;
 100                         char *chandle;
 101                         CURLcode sta;
 102                         CURLMcode msta;
 103                         AsyncIO  *IO;
 104
 105                         chandle = NULL;;
 106                         chnd = msg->easy_handle;
 107                         sta = curl_easy_getinfo(chnd, CURLINFO_PRIVATE, &chandle);
 108                         syslog(LOG_ERR, "EVCURL: request complete\n");
 109                         if (sta)
 110                                 syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
 111                         IO = (AsyncIO *)chandle;
 112
 113                         ev_io_stop(event_base, &IO->recv_event);
 114                         ev_io_stop(event_base, &IO->send_event);
 115
 116                         sta = msg->data.result;
 117                         if (sta) {
 118                                 EV_syslog(LOG_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc);
 119                                 EV_syslog(LOG_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
 120                         }
 121                         sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode);
 122                         if (sta)
 123                                 EV_syslog(LOG_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
 124                         EV_syslog(LOG_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode);
 125
 126
 127                         curl_slist_free_all(IO->HttpReq.headers);
 128                         msta = curl_multi_remove_handle(global.mhnd, chnd);
 129                         if (msta)
 130                                 EV_syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 131
 132                         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 133
 134                         IO->HttpReq.attached = 0;
 135                         switch(IO->SendDone(IO))
 136                         {
 137                         case eDBQuery:
 138                                 curl_easy_cleanup(IO->HttpReq.chnd);
 139                                 IO->HttpReq.chnd = NULL;
 140                                 break;
 141                         case eSendDNSQuery:
 142                         case eReadDNSReply:
 143                         case eConnect:
 144                         case eSendReply:
 145                         case eSendMore:
 146                         case eSendFile:
 147                         case eReadMessage:
 148                         case eReadMore:
 149                         case eReadPayload:
 150                         case eReadFile:
 151                                 curl_easy_cleanup(IO->HttpReq.chnd);
 152                                 IO->HttpReq.chnd = NULL;
 153                                 break;
 154                         case eTerminateConnection:
 155                         case eAbort:
 156                                 curl_easy_cleanup(IO->HttpReq.chnd);
 157                                 IO->HttpReq.chnd = NULL;
 158                                 FreeStrBuf(&IO->HttpReq.ReplyData);
 159                                 FreeURL(&IO->ConnectMe);
 160                                 RemoveContext(IO->CitContext);
 161                                 IO->Terminate(IO);
 162                         }
 163                 }
 164         }
 165 }
 166
 167 static void
 168 stepmulti(void *data, curl_socket_t fd, int which)
 169 {
 170         int running_handles = 0;
 171         CURLMcode msta;
 172
 173         msta = curl_multi_socket_action(global.mhnd, fd, which, &running_handles);
 174         syslog(LOG_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
 175         if (msta)
 176                 syslog(LOG_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta));
 177         if (global.nrun != running_handles)
 178                 gotstatus(running_handles);
 179 }
 180
 181 static void
 182 gottime(struct ev_loop *loop, ev_timer *timeev, int events) {
 183         syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n");
 184         stepmulti(NULL, CURL_SOCKET_TIMEOUT, 0);
 185 }
 186
 187 static void
 188 got_in(struct ev_loop *loop, ev_io *ioev, int events) {
 189         syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
 190         stepmulti(ioev->data, ioev->fd, CURL_CSELECT_IN);
 191 }
 192
 193 static void
 194 got_out(struct ev_loop *loop, ev_io *ioev, int events) {
 195         syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
 196         stepmulti(ioev->data, ioev->fd, CURL_CSELECT_OUT);
 197 }
 198
 199 static size_t
 200 gotdata(void *data, size_t size, size_t nmemb, void *cglobal) {
 201         AsyncIO *IO = (AsyncIO*) cglobal;
 202
 203         if (IO->HttpReq.ReplyData == NULL)
 204         {
 205             IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
 206         }
 207         return CurlFillStrBuf_callback(data, size, nmemb, IO->HttpReq.ReplyData);
 208 }
 209
 210 static int
 211 gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
 212         syslog(LOG_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
 213         evcurl_global_data *global = cglobal;
 214         ev_timer_stop(EV_DEFAULT, &global->timeev);
 215         if (tblock_ms < 0 || 14000 < tblock_ms)
 216                 tblock_ms = 14000;
 217         ev_timer_set(&global->timeev, 0.5e-3 + 1.0e-3 * tblock_ms, 14.0);
 218         ev_timer_start(EV_DEFAULT_UC, &global->timeev);
 219         curl_multi_perform(global, &global->nrun);
 220         return 0;
 221 }
 222
 223 static int
 224 gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *vIO) {
 225         evcurl_global_data *global = cglobal;
 226         CURLM *mhnd = global->mhnd;
 227         char *f;
 228         AsyncIO *IO = (AsyncIO*) vIO;
 229         CURLcode sta;
 230         const char *Action;
 231
 232         if (IO == NULL) {
 233                 sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f);
 234                 if (sta) {
 235                         EV_syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
 236                         return -1;
 237                 }
 238                 IO = (AsyncIO *) f;
 239                 EV_syslog(LOG_DEBUG, "EVCURL: got socket for URL: %s\n", IO->ConnectMe->PlainUrl);
 240                 if (IO->SendBuf.fd != 0)
 241                 {
 242                         ev_io_stop(event_base, &IO->recv_event);
 243                         ev_io_stop(event_base, &IO->send_event);
 244                 }
 245                 IO->SendBuf.fd = fd;
 246                 ev_io_init(&IO->recv_event, &got_in, fd, EV_READ);
 247                 ev_io_init(&IO->send_event, &got_out, fd, EV_WRITE);
 248                 curl_multi_assign(mhnd, fd, IO);
 249         }
 250
 251         Action = "";
 252         switch (action)
 253         {
 254         case CURL_POLL_NONE:
 255                 Action = "CURL_POLL_NONE";
 256                 break;
 257         case CURL_POLL_REMOVE:
 258                 Action = "CURL_POLL_REMOVE";
 259                 break;
 260         case CURL_POLL_IN:
 261                 Action = "CURL_POLL_IN";
 262                 break;
 263         case CURL_POLL_OUT:
 264                 Action = "CURL_POLL_OUT";
 265                 break;
 266         case CURL_POLL_INOUT:
 267                 Action = "CURL_POLL_INOUT";
 268                 break;
 269         }
 270
 271
 272         EV_syslog(LOG_DEBUG, "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n", (int)fd, Action, action);
 273
 274         switch (action)
 275         {
 276         case CURL_POLL_NONE:
 277                 EVM_syslog(LOG_ERR,"EVCURL: called first time to register this sockwatcker\n");
 278                 break;
 279         case CURL_POLL_REMOVE:
 280                 EVM_syslog(LOG_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
 281                 ev_io_stop(event_base, &IO->recv_event);
 282                 ev_io_stop(event_base, &IO->send_event);
 283                 break;
 284         case CURL_POLL_IN:
 285                 ev_io_start(event_base, &IO->recv_event);
 286                 ev_io_stop(event_base, &IO->send_event);
 287                 break;
 288         case CURL_POLL_OUT:
 289                 ev_io_start(event_base, &IO->send_event);
 290                 ev_io_stop(event_base, &IO->recv_event);
 291                 break;
 292         case CURL_POLL_INOUT:
 293                 ev_io_start(event_base, &IO->send_event);
 294                 ev_io_start(event_base, &IO->recv_event);
 295                 break;
 296         }
 297         return 0;
 298 }
 299
 300 void curl_init_connectionpool(void)
 301 {
 302         CURLM *mhnd ;
 303
 304         ev_timer_init(&global.timeev, &gottime, 14.0, 14.0);
 305         global.timeev.data = (void *)&global;
 306         global.nrun = -1;
 307         CURLcode sta = curl_global_init(CURL_GLOBAL_ALL);
 308
 309         if (sta)
 310         {
 311                 syslog(LOG_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta));
 312                 exit(1);
 313         }
 314         mhnd = global.mhnd = curl_multi_init();
 315         if (!mhnd)
 316         {
 317                 syslog(LOG_ERR,"EVCURL: error initializing curl multi handle\n");
 318                 exit(3);
 319         }
 320
 321         MOPT(SOCKETFUNCTION, &gotwatchsock);
 322         MOPT(SOCKETDATA, (void *)&global);
 323         MOPT(TIMERFUNCTION, &gotwatchtime);
 324         MOPT(TIMERDATA, (void *)&global);
 325
 326         return;
 327 }
 328
 329 int evcurl_init(AsyncIO *IO)
 330 {
 331         CURLcode sta;
 332         CURL *chnd;
 333
 334         EVM_syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
 335         IO->HttpReq.attached = 0;
 336         chnd = IO->HttpReq.chnd = curl_easy_init();
 337         if (!chnd)
 338         {
 339                 EVM_syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
 340                 return 0;
 341         }
 342
 343         OPT(VERBOSE, (long)1);
 344                 /* unset in production */
 345         OPT(NOPROGRESS, 1L);
 346         OPT(NOSIGNAL, 1L);
 347         OPT(FAILONERROR, (long)1);
 348         OPT(ENCODING, "");
 349         OPT(FOLLOWLOCATION, (long)0);
 350         OPT(MAXREDIRS, (long)0);
 351         OPT(USERAGENT, CITADEL);
 352
 353         OPT(TIMEOUT, (long)1800);
 354         OPT(LOW_SPEED_LIMIT, (long)64);
 355         OPT(LOW_SPEED_TIME, (long)600);
 356         OPT(CONNECTTIMEOUT, (long)600);
 357         OPT(PRIVATE, (void *)IO);
 358
 359         OPT(FORBID_REUSE, 1);
 360         OPT(WRITEFUNCTION, &gotdata);
 361         OPT(WRITEDATA, (void *)IO);
 362         OPT(ERRORBUFFER, IO->HttpReq.errdesc);
 363
 364         if (
 365                 (!IsEmptyStr(config.c_ip_addr))
 366                 && (strcmp(config.c_ip_addr, "*"))
 367                 && (strcmp(config.c_ip_addr, "::"))
 368                 && (strcmp(config.c_ip_addr, "0.0.0.0"))
 369         ) {
 370                 OPT(INTERFACE, config.c_ip_addr);
 371         }
 372
 373 #ifdef CURLOPT_HTTP_CONTENT_DECODING
 374         OPT(HTTP_CONTENT_DECODING, 1);
 375         OPT(ENCODING, "");
 376 #endif
 377
 378         IO->HttpReq.headers = curl_slist_append(IO->HttpReq.headers, "Connection: close");
 379
 380         return 1;
 381 }
 382
 383
 384 static void IOcurl_abort_shutdown_callback(struct ev_loop *loop, ev_cleanup *watcher, int revents)
 385 {
 386         CURLMcode msta;
 387         AsyncIO *IO = watcher->data;
 388         EV_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__);
 389
 390         curl_slist_free_all(IO->HttpReq.headers);
 391         msta = curl_multi_remove_handle(global.mhnd, IO->HttpReq.chnd);
 392         if (msta)
 393                 EV_syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 394
 395         curl_easy_cleanup(IO->HttpReq.chnd);
 396         IO->HttpReq.chnd = NULL;
 397         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 398         ev_io_stop(event_base, &IO->recv_event);
 399         ev_io_stop(event_base, &IO->send_event);
 400         assert(IO->ShutdownAbort);
 401         IO->ShutdownAbort(IO);
 402 }
 403 eNextState
 404 evcurl_handle_start(AsyncIO *IO)
 405 {
 406         CURLMcode msta;
 407         CURLcode sta;
 408         CURL *chnd;
 409
 410         chnd = IO->HttpReq.chnd;
 411         EV_syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
 412         OPT(URL, IO->ConnectMe->PlainUrl);
 413         if (StrLength(IO->ConnectMe->CurlCreds))
 414         {
 415                 OPT(HTTPAUTH, (long)CURLAUTH_BASIC);
 416                 OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds));
 417         }
 418         if (StrLength(IO->HttpReq.PostData) > 0)
 419         {
 420                 OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData));
 421                 OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData));
 422
 423         }
 424         else if ((IO->HttpReq.PlainPostDataLen != 0) && (IO->HttpReq.PlainPostData != NULL))
 425         {
 426                 OPT(POSTFIELDS, IO->HttpReq.PlainPostData);
 427                 OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen);
 428         }
 429         OPT(HTTPHEADER, IO->HttpReq.headers);
 430
 431         IO->NextState = eConnect;
 432         EVM_syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
 433         msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
 434         if (msta)
 435                 EV_syslog(LOG_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
 436         IO->HttpReq.attached = 1;
 437         ev_async_send (event_base, &WakeupCurl);
 438         ev_cleanup_init(&IO->abort_by_shutdown,
 439                         IOcurl_abort_shutdown_callback);
 440         ev_cleanup_start(event_base, &IO->abort_by_shutdown);
 441         return eReadMessage;
 442 }
 443
 444 static void WakeupCurlCallback(EV_P_ ev_async *w, int revents)
 445 {
 446         syslog(LOG_DEBUG, "EVCURL: waking up curl multi handle\n");
 447
 448         curl_multi_perform(&global, CURL_POLL_NONE);
 449 }
 450
 451 static void evcurl_shutdown (void)
 452 {
 453         curl_global_cleanup();
 454         curl_multi_cleanup(global.mhnd);
 455         syslog(LOG_DEBUG, "client_event_thread() initializing\n");
 456 }
 457 /*****************************************************************************
 458  *                       libevent integration                                *
 459  *****************************************************************************/
 460 /*
 461  * client event queue plus its methods.
 462  * this currently is the main loop (which may change in some future?)
 463  */
 464 int evbase_count = 0;
 465 int event_add_pipe[2] = {-1, -1};
 466 pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
 467 HashList *QueueEvents = NULL;
 468 HashList *InboundEventQueue = NULL;
 469 HashList *InboundEventQueues[2] = { NULL, NULL };
 470
 471 ev_async AddJob;
 472 ev_async ExitEventLoop;
 473
 474 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 475 {
 476         HashList *q;
 477         void *v;
 478         HashPos  *It;
 479         long len;
 480         const char *Key;
 481
 482         /* get the control command... */
 483         pthread_mutex_lock(&EventQueueMutex);
 484
 485         if (InboundEventQueues[0] == InboundEventQueue) {
 486                 InboundEventQueue = InboundEventQueues[1];
 487                 q = InboundEventQueues[0];
 488         }
 489         else {
 490                 InboundEventQueue = InboundEventQueues[0];
 491                 q = InboundEventQueues[1];
 492         }
 493         pthread_mutex_unlock(&EventQueueMutex);
 494
 495         It = GetNewHashPos(q, 0);
 496         while (GetNextHashPos(q, It, &len, &Key, &v))
 497         {
 498                 IOAddHandler *h = v;
 499                 if (h->IO->ID == 0)
 500                         h->IO->ID = EvIDSource++;
 501                 h->EvAttch(h->IO);
 502         }
 503         DeleteHashPos(&It);
 504         DeleteHashContent(&q);
 505         syslog(LOG_DEBUG, "EVENT Q Add done.\n");
 506 }
 507
 508
 509 static void EventExitCallback(EV_P_ ev_async *w, int revents)
 510 {
 511         ev_break(event_base, EVBREAK_ALL);
 512
 513         syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 514 }
 515
 516
 517
 518 void InitEventQueue(void)
 519 {
 520         struct rlimit LimitSet;
 521
 522         pthread_mutex_init(&EventQueueMutex, NULL);
 523
 524         if (pipe(event_add_pipe) != 0) {
 525                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 526                 abort();
 527         }
 528         LimitSet.rlim_cur = 1;
 529         LimitSet.rlim_max = 1;
 530         setrlimit(event_add_pipe[1], &LimitSet);
 531
 532         QueueEvents = NewHash(1, Flathash);
 533         InboundEventQueues[0] = NewHash(1, Flathash);
 534         InboundEventQueues[1] = NewHash(1, Flathash);
 535         InboundEventQueue = InboundEventQueues[0];
 536 }
 537 /*
 538  * this thread operates the select() etc. via libev.
 539  *
 540  *
 541  */
 542 void *client_event_thread(void *arg)
 543 {
 544         struct CitContext libev_client_CC;
 545
 546         CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
 547 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 548         syslog(LOG_DEBUG, "client_event_thread() initializing\n");
 549
 550         event_base = ev_default_loop (EVFLAG_AUTO);
 551         ev_async_init(&AddJob, QueueEventAddCallback);
 552         ev_async_start(event_base, &AddJob);
 553         ev_async_init(&ExitEventLoop, EventExitCallback);
 554         ev_async_start(event_base, &ExitEventLoop);
 555         ev_async_init(&WakeupCurl, WakeupCurlCallback);
 556         ev_async_start(event_base, &WakeupCurl);
 557
 558         curl_init_connectionpool();
 559
 560         ev_run (event_base, 0);
 561
 562         syslog(LOG_DEBUG, "client_event_thread() exiting\n");
 563
 564 ///what todo here?      CtdlClearSystemContext();
 565         ev_loop_destroy (EV_DEFAULT_UC);
 566         DeleteHash(&QueueEvents);
 567         InboundEventQueue = NULL;
 568         DeleteHash(&InboundEventQueues[0]);
 569         DeleteHash(&InboundEventQueues[1]);
 570 /*      citthread_mutex_destroy(&EventQueueMutex); TODO */
 571         evcurl_shutdown();
 572
 573         return(NULL);
 574 }
 575 /*------------------------------------------------------------------------------*/
 576 /*
 577  * DB-Queue; does async bdb operations.
 578  * has its own set of handlers.
 579  */
 580 ev_loop *event_db;
 581 int evdb_count = 0;
 582 int evdb_add_pipe[2] = {-1, -1};
 583 pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
 584 HashList *DBQueueEvents = NULL;
 585 HashList *DBInboundEventQueue = NULL;
 586 HashList *DBInboundEventQueues[2] = { NULL, NULL };
 587
 588 ev_async DBAddJob;
 589 ev_async DBExitEventLoop;
 590
 591 extern void ShutDownDBCLient(AsyncIO *IO);
 592
 593 static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
 594 {
 595         HashList *q;
 596         void *v;
 597         HashPos  *It;
 598         long len;
 599         const char *Key;
 600
 601         /* get the control command... */
 602         pthread_mutex_lock(&DBEventQueueMutex);
 603
 604         if (DBInboundEventQueues[0] == DBInboundEventQueue) {
 605                 DBInboundEventQueue = DBInboundEventQueues[1];
 606                 q = DBInboundEventQueues[0];
 607         }
 608         else {
 609                 DBInboundEventQueue = DBInboundEventQueues[0];
 610                 q = DBInboundEventQueues[1];
 611         }
 612         pthread_mutex_unlock(&DBEventQueueMutex);
 613
 614         It = GetNewHashPos(q, 0);
 615         while (GetNextHashPos(q, It, &len, &Key, &v))
 616         {
 617                 IOAddHandler *h = v;
 618                 eNextState rc;
 619                 if (h->IO->ID == 0)
 620                         h->IO->ID = EvIDSource++;
 621                 rc = h->EvAttch(h->IO);
 622                 switch (rc)
 623                 {
 624                 case eAbort:
 625                     ShutDownDBCLient(h->IO);
 626                 default:
 627                     break;
 628                 }
 629         }
 630         DeleteHashPos(&It);
 631         DeleteHashContent(&q);
 632         syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
 633 }
 634
 635
 636 static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
 637 {
 638         syslog(LOG_DEBUG, "DB EVENT Q exiting.\n");
 639         ev_break(event_db, EVBREAK_ALL);
 640 }
 641
 642
 643
 644 void DBInitEventQueue(void)
 645 {
 646         struct rlimit LimitSet;
 647
 648         pthread_mutex_init(&DBEventQueueMutex, NULL);
 649
 650         if (pipe(evdb_add_pipe) != 0) {
 651                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 652                 abort();
 653         }
 654         LimitSet.rlim_cur = 1;
 655         LimitSet.rlim_max = 1;
 656         setrlimit(evdb_add_pipe[1], &LimitSet);
 657
 658         DBQueueEvents = NewHash(1, Flathash);
 659         DBInboundEventQueues[0] = NewHash(1, Flathash);
 660         DBInboundEventQueues[1] = NewHash(1, Flathash);
 661         DBInboundEventQueue = DBInboundEventQueues[0];
 662 }
 663
 664 /*
 665  * this thread operates writing to the message database via libev.
 666  *
 667  *
 668  */
 669 void *db_event_thread(void *arg)
 670 {
 671         struct CitContext libev_msg_CC;
 672
 673         CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
 674 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 675         syslog(LOG_DEBUG, "dbevent_thread() initializing\n");
 676
 677         event_db = ev_loop_new (EVFLAG_AUTO);
 678
 679         ev_async_init(&DBAddJob, DBQueueEventAddCallback);
 680         ev_async_start(event_db, &DBAddJob);
 681         ev_async_init(&DBExitEventLoop, DBEventExitCallback);
 682         ev_async_start(event_db, &DBExitEventLoop);
 683
 684         ev_run (event_db, 0);
 685
 686         syslog(LOG_DEBUG, "dbevent_thread() exiting\n");
 687
 688 //// what to do here?   CtdlClearSystemContext();
 689         ev_loop_destroy (event_db);
 690
 691         DeleteHash(&DBQueueEvents);
 692         DBInboundEventQueue = NULL;
 693         DeleteHash(&DBInboundEventQueues[0]);
 694         DeleteHash(&DBInboundEventQueues[1]);
 695 /*      citthread_mutex_destroy(&DBEventQueueMutex); TODO */
 696
 697         return(NULL);
 698 }
 699
 700 void ShutDownEventQueues(void)
 701 {
 702         syslog(LOG_DEBUG, "EVENT Qs triggering exits.\n");
 703
 704         pthread_mutex_lock(&DBEventQueueMutex);
 705         ev_async_send (event_db, &DBExitEventLoop);
 706         pthread_mutex_unlock(&DBEventQueueMutex);
 707
 708         pthread_mutex_lock(&EventQueueMutex);
 709         ev_async_send (EV_DEFAULT_ &ExitEventLoop);
 710         pthread_mutex_unlock(&EventQueueMutex);
 711 }
 712
 713 CTDL_MODULE_INIT(event_client)
 714 {
 715         if (!threading)
 716         {
 717                 CtdlRegisterCleanupHook(ShutDownEventQueues);
 718                 InitEventQueue();
 719                 DBInitEventQueue();
 720                 CtdlThreadCreate(/*"Client event", */ client_event_thread);
 721                 CtdlThreadCreate(/*"DB event", */db_event_thread);
 722 /// todo register shutdown callback.
 723         }
 724         return "event";
 725 }