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 <arpa/inet.h>
  48 #include <libcitadel.h>
  49 #include <curl/curl.h>
  50 #include <curl/multi.h>
  51 #include "citadel.h"
  52 #include "server.h"
  53 #include "citserver.h"
  54 #include "support.h"
  55
  56 #include "ctdl_module.h"
  57
  58 #include "event_client.h"
  59 #include "serv_curl.h"
  60
  61 ev_loop *event_base;
  62
  63 /*****************************************************************************
  64  *                   libevent / curl integration                             *
  65  *****************************************************************************/
  66 #define MOPT(s, v)                                                      \
  67         do {                                                            \
  68                 sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
  69                 if (sta) {                                              \
  70                         syslog(LOG_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \
  71                         exit (1);                                       \
  72                 }                                                       \
  73         } while (0)
  74
  75 typedef struct _evcurl_global_data {
  76         int magic;
  77         CURLM *mhnd;
  78         ev_timer timeev;
  79         int nrun;
  80 } evcurl_global_data;
  81
  82 ev_async WakeupCurl;
  83 evcurl_global_data global;
  84
  85 static void
  86 gotstatus(evcurl_global_data *global, int nnrun)
  87 {
  88         CURLM *mhnd;
  89         CURLMsg *msg;
  90         int nmsg;
  91
  92         global->nrun = nnrun;
  93         mhnd = global->mhnd;
  94
  95         syslog(LOG_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
  96         while ((msg = curl_multi_info_read(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
 115                         sta = msg->data.result;
 116                         if (sta) {
 117                                 syslog(LOG_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc);
 118                                 syslog(LOG_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
 119                         }
 120                         sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode);
 121                         if (sta)
 122                                 syslog(LOG_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
 123                         syslog(LOG_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode);
 124 ///                     syslog(LOG_ERR, "EVCURL: disconnecting [%s]\nn", ChrPtr(IO->ConnectMe->URL));
 125                         msta = curl_multi_remove_handle(mhnd, chnd);
 126                         if (msta)
 127                                 syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 128 ///                     syslog(LOG_ERR, "EVCURL: disconnection done [%s]\nn", ChrPtr(IO->ConnectMe->URL));
 129
 130                         IO->HttpReq.attached = 0;
 131                         IO->SendDone(IO);
 132 ////                    syslog(LOG_ERR, "EVCURL: done with [%s]\nn", ChrPtr(IO->ConnectMe->URL));
 133
 134
 135                         curl_easy_cleanup(IO->HttpReq.chnd);
 136                         curl_slist_free_all(IO->HttpReq.headers);
 137 /*
 138                         FreeStrBuf(&IO->HttpReq.ReplyData);
 139                         FreeURL(&IO->ConnectMe);
 140                         RemoveContext(IO->CitContext);
 141                         IO->Terminate(IO);
 142 */
 143                 }
 144         }
 145 }
 146
 147 static void
 148 stepmulti(evcurl_global_data *global, curl_socket_t fd) {
 149         int nnrun;
 150         CURLMcode msta;
 151
 152         if (global == NULL) {
 153             syslog(LOG_DEBUG, "EVCURL: stepmulti(NULL): wtf?\n");
 154             return;
 155         }
 156         msta = curl_multi_socket_action(global->mhnd, fd, 0, &nnrun);
 157         syslog(LOG_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
 158         if (msta)
 159                 syslog(LOG_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta));
 160         if (global->nrun != nnrun)
 161                 gotstatus(global, nnrun);
 162 }
 163
 164 static void
 165 gottime(struct ev_loop *loop, ev_timer *timeev, int events) {
 166         syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n");
 167         evcurl_global_data *global = (void *)timeev->data;
 168         stepmulti(global, CURL_SOCKET_TIMEOUT);
 169 }
 170
 171 static void
 172 gotio(struct ev_loop *loop, ev_io *ioev, int events) {
 173         syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
 174         stepmulti(&global, ioev->fd);
 175 }
 176
 177 static size_t
 178 gotdata(void *data, size_t size, size_t nmemb, void *cglobal) {
 179         AsyncIO *IO = (AsyncIO*) cglobal;
 180         //evcurl_request_data *D = (evcurl_request_data*) data;
 181 ///        syslog(LOG_DEBUG, "EVCURL: gotdata(): calling CurlFillStrBuf_callback()\n");
 182
 183         if (IO->HttpReq.ReplyData == NULL)
 184         {
 185             IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
 186         }
 187         return CurlFillStrBuf_callback(data, size, nmemb, IO->HttpReq.ReplyData);
 188 }
 189
 190 static int
 191 gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
 192         syslog(LOG_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
 193         evcurl_global_data *global = cglobal;
 194         ev_timer_stop(EV_DEFAULT, &global->timeev);
 195         if (tblock_ms < 0 || 14000 < tblock_ms)
 196                 tblock_ms = 14000;
 197         ev_timer_set(&global->timeev, 0.5e-3 + 1.0e-3 * tblock_ms, 14.0);
 198         ev_timer_start(EV_DEFAULT_UC, &global->timeev);
 199         curl_multi_perform(global, CURL_POLL_NONE);
 200         return 0;
 201 }
 202
 203 static int
 204 gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *vio) {
 205         evcurl_global_data *global = cglobal;
 206         CURLM *mhnd = global->mhnd;
 207         char *f;
 208         AsyncIO *IO = (AsyncIO*) vio;
 209         CURLcode sta;
 210
 211         syslog(LOG_DEBUG, "EVCURL: gotwatchsock called fd=%d action=%d\n", (int)fd, action);
 212
 213         if (IO == NULL) {
 214                 syslog(LOG_ERR,"EVCURL: called first time to register this sockwatcker\n");
 215                 sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f);
 216                 if (sta) {
 217                         syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
 218                         return -1;
 219                 }
 220                 IO = (AsyncIO *) f;
 221                 ev_init(&IO->recv_event, &gotio);
 222                 curl_multi_assign(mhnd, fd, IO);
 223         }
 224         if (action == CURL_POLL_REMOVE) {
 225                 syslog(LOG_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
 226                 ev_io_stop(event_base, &IO->recv_event);
 227                 FreeStrBuf(&IO->HttpReq.ReplyData);
 228                 FreeURL(&IO->ConnectMe);
 229                 RemoveContext(IO->CitContext);
 230                 IO->Terminate(IO);
 231         } else {
 232                 int events = (action & CURL_POLL_IN ? EV_READ : 0) | (action & CURL_POLL_OUT ? EV_WRITE : 0);
 233                 ev_io_stop(event_base, &IO->recv_event);
 234                 if (events) {
 235                         ev_io_set(&IO->recv_event, fd, events);
 236                         ev_io_start(event_base, &IO->recv_event);
 237                 }
 238         }
 239         return 0;
 240 }
 241
 242 void curl_init_connectionpool(void)
 243 {
 244         CURLM *mhnd ;
 245
 246         ev_timer_init(&global.timeev, &gottime, 14.0, 14.0);
 247         global.timeev.data = (void *)&global;
 248         global.nrun = -1;
 249         CURLcode sta = curl_global_init(CURL_GLOBAL_ALL);
 250
 251         if (sta)
 252         {
 253                 syslog(LOG_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta));
 254                 exit(1);
 255         }
 256         mhnd = global.mhnd = curl_multi_init();
 257         if (!mhnd)
 258         {
 259                 syslog(LOG_ERR,"EVCURL: error initializing curl multi handle\n");
 260                 exit(3);
 261         }
 262
 263         MOPT(SOCKETFUNCTION, &gotwatchsock);
 264         MOPT(SOCKETDATA, (void *)&global);
 265         MOPT(TIMERFUNCTION, &gotwatchtime);
 266         MOPT(TIMERDATA, (void *)&global);
 267
 268         return;
 269 }
 270
 271
 272
 273
 274 int evcurl_init(AsyncIO *IO,
 275                 void *CustomData,
 276                 const char* Desc,
 277                 IO_CallBack CallBack,
 278                 IO_CallBack Terminate)
 279 {
 280         CURLcode sta;
 281         CURL *chnd;
 282
 283         syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
 284         IO->HttpReq.attached = 0;
 285         IO->SendDone = CallBack;
 286         IO->Terminate = Terminate;
 287         chnd = IO->HttpReq.chnd = curl_easy_init();
 288         if (!chnd)
 289         {
 290                 syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
 291                 return 1;
 292         }
 293
 294         strcpy(IO->HttpReq.errdesc, Desc);
 295
 296         OPT(VERBOSE, (long)1);
 297                 /* unset in production */
 298         OPT(NOPROGRESS, (long)1);
 299         OPT(NOSIGNAL, (long)1);
 300         OPT(FAILONERROR, (long)1);
 301         OPT(ENCODING, "");
 302         OPT(FOLLOWLOCATION, (long)1);
 303         OPT(MAXREDIRS, (long)7);
 304         OPT(USERAGENT, CITADEL);
 305
 306         OPT(TIMEOUT, (long)1800);
 307         OPT(LOW_SPEED_LIMIT, (long)64);
 308         OPT(LOW_SPEED_TIME, (long)600);
 309         OPT(CONNECTTIMEOUT, (long)600);
 310         OPT(PRIVATE, (void *)IO);
 311
 312         OPT(FORBID_REUSE, 1);
 313         OPT(WRITEFUNCTION, &gotdata);
 314         OPT(WRITEDATA, (void *)IO);
 315         OPT(ERRORBUFFER, IO->HttpReq.errdesc);
 316
 317         if (
 318                 (!IsEmptyStr(config.c_ip_addr))
 319                 && (strcmp(config.c_ip_addr, "*"))
 320                 && (strcmp(config.c_ip_addr, "::"))
 321                 && (strcmp(config.c_ip_addr, "0.0.0.0"))
 322         ) {
 323                 OPT(INTERFACE, config.c_ip_addr);
 324         }
 325                 /* point to a structure that points back to the perl structure and stuff */
 326         syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
 327         OPT(URL, IO->ConnectMe->PlainUrl);
 328         if (StrLength(IO->ConnectMe->CurlCreds))
 329         {
 330                 OPT(HTTPAUTH, (long)CURLAUTH_BASIC);
 331                 OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds));
 332         }
 333 #ifdef CURLOPT_HTTP_CONTENT_DECODING
 334         OPT(HTTP_CONTENT_DECODING, 1);
 335         OPT(ENCODING, "");
 336 #endif
 337         if (StrLength(IO->HttpReq.PostData) > 0)
 338         {
 339                 OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData));
 340                 OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData));
 341
 342         }
 343         else if ((IO->HttpReq.PlainPostDataLen != 0) && (IO->HttpReq.PlainPostData != NULL))
 344         {
 345                 OPT(POSTFIELDS, IO->HttpReq.PlainPostData);
 346                 OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen);
 347         }
 348
 349         IO->HttpReq.headers = curl_slist_append(IO->HttpReq.headers, "Connection: close");
 350         OPT(HTTPHEADER, IO->HttpReq.headers);
 351
 352         return 1;
 353 }
 354
 355 eNextState
 356 evcurl_handle_start(AsyncIO *IO)
 357 {
 358         CURLMcode msta;
 359
 360         syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
 361         msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
 362         if (msta)
 363                 syslog(LOG_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
 364         IO->HttpReq.attached = 1;
 365         ev_async_send (event_base, &WakeupCurl);
 366         return eReadMessage;
 367 }
 368
 369 static void WakeupCurlCallback(EV_P_ ev_async *w, int revents)
 370 {
 371         syslog(LOG_DEBUG, "EVCURL: waking up curl multi handle\n");
 372
 373         curl_multi_perform(&global, CURL_POLL_NONE);
 374 }
 375
 376 static void evcurl_shutdown (void)
 377 {
 378         curl_multi_cleanup(global.mhnd);
 379 }
 380 /*****************************************************************************
 381  *                       libevent integration                                *
 382  *****************************************************************************/
 383 /*
 384  * client event queue plus its methods.
 385  * this currently is the main loop (which may change in some future?)
 386  */
 387 int evbase_count = 0;
 388 int event_add_pipe[2] = {-1, -1};
 389 pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
 390 HashList *QueueEvents = NULL;
 391 HashList *InboundEventQueue = NULL;
 392 HashList *InboundEventQueues[2] = { NULL, NULL };
 393
 394 ev_async AddJob;
 395 ev_async ExitEventLoop;
 396
 397 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 398 {
 399         HashList *q;
 400         void *v;
 401         HashPos  *It;
 402         long len;
 403         const char *Key;
 404
 405         /* get the control command... */
 406         pthread_mutex_lock(&EventQueueMutex);
 407
 408         if (InboundEventQueues[0] == InboundEventQueue) {
 409                 InboundEventQueue = InboundEventQueues[1];
 410                 q = InboundEventQueues[0];
 411         }
 412         else {
 413                 InboundEventQueue = InboundEventQueues[0];
 414                 q = InboundEventQueues[1];
 415         }
 416         pthread_mutex_unlock(&EventQueueMutex);
 417
 418         It = GetNewHashPos(q, 0);
 419         while (GetNextHashPos(q, It, &len, &Key, &v))
 420         {
 421                 IOAddHandler *h = v;
 422                 h->EvAttch(h->IO);
 423         }
 424         DeleteHashPos(&It);
 425         DeleteHashContent(&q);
 426         syslog(LOG_DEBUG, "EVENT Q Read done.\n");
 427 }
 428
 429
 430 static void EventExitCallback(EV_P_ ev_async *w, int revents)
 431 {
 432         ev_break(event_base, EVBREAK_ALL);
 433
 434         syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 435 }
 436
 437
 438
 439 void InitEventQueue(void)
 440 {
 441         struct rlimit LimitSet;
 442
 443         pthread_mutex_init(&EventQueueMutex, NULL);
 444
 445         if (pipe(event_add_pipe) != 0) {
 446                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 447                 abort();
 448         }
 449         LimitSet.rlim_cur = 1;
 450         LimitSet.rlim_max = 1;
 451         setrlimit(event_add_pipe[1], &LimitSet);
 452
 453         QueueEvents = NewHash(1, Flathash);
 454         InboundEventQueues[0] = NewHash(1, Flathash);
 455         InboundEventQueues[1] = NewHash(1, Flathash);
 456         InboundEventQueue = InboundEventQueues[0];
 457 }
 458 /*
 459  * this thread operates the select() etc. via libev.
 460  *
 461  *
 462  */
 463 void *client_event_thread(void *arg)
 464 {
 465         struct CitContext libev_client_CC;
 466
 467         CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
 468 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 469         syslog(LOG_DEBUG, "client_ev_thread() initializing\n");
 470
 471         event_base = ev_default_loop (EVFLAG_AUTO);
 472
 473         ev_async_init(&AddJob, QueueEventAddCallback);
 474         ev_async_start(event_base, &AddJob);
 475         ev_async_init(&ExitEventLoop, EventExitCallback);
 476         ev_async_start(event_base, &ExitEventLoop);
 477         ev_async_init(&WakeupCurl, WakeupCurlCallback);
 478         ev_async_start(event_base, &WakeupCurl);
 479
 480         curl_init_connectionpool();
 481
 482         ev_run (event_base, 0);
 483
 484
 485 ///what todo here?      CtdlClearSystemContext();
 486         ev_loop_destroy (EV_DEFAULT_UC);
 487
 488         DeleteHash(&QueueEvents);
 489         InboundEventQueue = NULL;
 490         DeleteHash(&InboundEventQueues[0]);
 491         DeleteHash(&InboundEventQueues[1]);
 492 /*      citthread_mutex_destroy(&EventQueueMutex); TODO */
 493         evcurl_shutdown();
 494
 495         return(NULL);
 496 }
 497 /*------------------------------------------------------------------------------*/
 498 /*
 499  * DB-Queue; does async bdb operations.
 500  * has its own set of handlers.
 501  */
 502 ev_loop *event_db;
 503 int evdb_count = 0;
 504 int evdb_add_pipe[2] = {-1, -1};
 505 pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
 506 HashList *DBQueueEvents = NULL;
 507 HashList *DBInboundEventQueue = NULL;
 508 HashList *DBInboundEventQueues[2] = { NULL, NULL };
 509
 510 ev_async DBAddJob;
 511 ev_async DBExitEventLoop;
 512
 513 extern void ShutDownDBCLient(AsyncIO *IO);
 514
 515 static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
 516 {
 517         HashList *q;
 518         void *v;
 519         HashPos  *It;
 520         long len;
 521         const char *Key;
 522
 523         /* get the control command... */
 524         pthread_mutex_lock(&DBEventQueueMutex);
 525
 526         if (DBInboundEventQueues[0] == DBInboundEventQueue) {
 527                 DBInboundEventQueue = DBInboundEventQueues[1];
 528                 q = DBInboundEventQueues[0];
 529         }
 530         else {
 531                 DBInboundEventQueue = DBInboundEventQueues[0];
 532                 q = DBInboundEventQueues[1];
 533         }
 534         pthread_mutex_unlock(&DBEventQueueMutex);
 535
 536         It = GetNewHashPos(q, 0);
 537         while (GetNextHashPos(q, It, &len, &Key, &v))
 538         {
 539                 IOAddHandler *h = v;
 540                 eNextState rc;
 541                 rc = h->EvAttch(h->IO);
 542                 switch (rc)
 543                 {
 544                 case eAbort:
 545                     ShutDownDBCLient(h->IO);
 546                 default:
 547                     break;
 548                 }
 549         }
 550         DeleteHashPos(&It);
 551         DeleteHashContent(&q);
 552         syslog(LOG_DEBUG, "DBEVENT Q Read done.\n");
 553 }
 554
 555
 556 static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
 557 {
 558         syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 559         ev_break(event_db, EVBREAK_ALL);
 560 }
 561
 562
 563
 564 void DBInitEventQueue(void)
 565 {
 566         struct rlimit LimitSet;
 567
 568         pthread_mutex_init(&DBEventQueueMutex, NULL);
 569
 570         if (pipe(evdb_add_pipe) != 0) {
 571                 syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 572                 abort();
 573         }
 574         LimitSet.rlim_cur = 1;
 575         LimitSet.rlim_max = 1;
 576         setrlimit(evdb_add_pipe[1], &LimitSet);
 577
 578         DBQueueEvents = NewHash(1, Flathash);
 579         DBInboundEventQueues[0] = NewHash(1, Flathash);
 580         DBInboundEventQueues[1] = NewHash(1, Flathash);
 581         DBInboundEventQueue = DBInboundEventQueues[0];
 582 }
 583
 584 /*
 585  * this thread operates writing to the message database via libev.
 586  *
 587  *
 588  */
 589 void *db_event_thread(void *arg)
 590 {
 591         struct CitContext libev_msg_CC;
 592
 593         CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
 594 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 595         syslog(LOG_DEBUG, "client_msgev_thread() initializing\n");
 596
 597         event_db = ev_loop_new (EVFLAG_AUTO);
 598
 599         ev_async_init(&DBAddJob, DBQueueEventAddCallback);
 600         ev_async_start(event_db, &DBAddJob);
 601         ev_async_init(&DBExitEventLoop, DBEventExitCallback);
 602         ev_async_start(event_db, &DBExitEventLoop);
 603
 604         ev_run (event_db, 0);
 605
 606
 607 //// what to do here?   CtdlClearSystemContext();
 608         ev_loop_destroy (event_db);
 609
 610         DeleteHash(&DBQueueEvents);
 611         DBInboundEventQueue = NULL;
 612         DeleteHash(&DBInboundEventQueues[0]);
 613         DeleteHash(&DBInboundEventQueues[1]);
 614 /*      citthread_mutex_destroy(&DBEventQueueMutex); TODO */
 615
 616         return(NULL);
 617 }
 618
 619 CTDL_MODULE_INIT(event_client)
 620 {
 621         if (!threading)
 622         {
 623                 InitEventQueue();
 624                 DBInitEventQueue();
 625                 CtdlThreadCreate(/*"Client event", */ client_event_thread);
 626                 CtdlThreadCreate(/*"DB event", */db_event_thread);
 627 /// todo register shutdown callback.
 628         }
 629         return "event";
 630 }