More places to remove the experimental #if
[citadel.git] / citadel / modules / eventclient / serv_eventclient.c
index 7b6a8c3dd27e8ca77ea6d3c668a617128c572b2e..06af6147647879ab460027923fa42908e5f726b6 100644 (file)
 
 #include "ctdl_module.h"
 
-#ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT
-
 #include "event_client.h"
 #include "serv_curl.h"
 
-int event_add_pipe[2] = {-1, -1};
-
-citthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
-HashList *QueueEvents = NULL;
-
-HashList *InboundEventQueue = NULL;
-HashList *InboundEventQueues[2] = { NULL, NULL };
-
-struct ev_loop *event_base;
-
-ev_async AddJob;   
-ev_async ExitEventLoop;
-ev_async WakeupCurl;
-
-extern struct ev_loop *event_base;
-
-void SockStateCb(void *data, int sock, int read, int write);
+ev_loop *event_base;
 
+/*****************************************************************************
+ *                   libevent / curl integration                             *
+ *****************************************************************************/
 #define MOPT(s, v)                                                     \
        do {                                                            \
                sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
                if (sta) {                                              \
-                       CtdlLogPrintf(CTDL_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \
+                       syslog(LOG_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \
                        exit (1);                                       \
                }                                                       \
        } while (0)
 
+typedef struct _evcurl_global_data {
+       int magic;
+       CURLM *mhnd;
+       ev_timer timeev;
+       int nrun;
+} evcurl_global_data;
 
+typedef struct _sockwatcher_data 
+{
+       evcurl_global_data *global;
+       ev_io ioev;
+} sockwatcher_data;
 
-/*****************************************************************************
- *                   libevent / curl integration                             *
- *****************************************************************************/
-
-
+ev_async WakeupCurl;
 evcurl_global_data global;
 
 static void
@@ -112,33 +104,45 @@ gotstatus(evcurl_global_data *global, int nnrun)
        global->nrun = nnrun;
        mhnd = global->mhnd;
 
-       CtdlLogPrintf(CTDL_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
+       syslog(LOG_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
        while ((msg = curl_multi_info_read(mhnd, &nmsg))) {
-               CtdlLogPrintf(CTDL_ERR, "EVCURL: got curl multi_info message msg=%d\n", msg->msg);
+               syslog(LOG_ERR, "EVCURL: got curl multi_info message msg=%d\n", msg->msg);
                if (CURLMSG_DONE == msg->msg) {
-                       CtdlLogPrintf(CTDL_ERR, "EVCURL: request complete\n");
-                       CURL *chnd = msg->easy_handle;
-                       char *chandle = NULL;;
-                       CURLcode sta = curl_easy_getinfo(chnd, CURLINFO_PRIVATE, &chandle);
+                       CURL *chnd;
+                       char *chandle;
+                       CURLcode sta;
+                       CURLMcode msta;
+                       AsyncIO  *IO;
+
+                       chandle = NULL;;
+                       chnd = msg->easy_handle;
+                       sta = curl_easy_getinfo(chnd, CURLINFO_PRIVATE, &chandle);
+                       syslog(LOG_ERR, "EVCURL: request complete\n");
                        if (sta)
-                               CtdlLogPrintf(CTDL_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
-                       evcurl_request_data  *handle = (void *)chandle;
-                       if (global != handle->global || chnd != handle->chnd)
-                               CtdlLogPrintf(CTDL_ERR, "EVCURL: internal evcurl error: unknown curl handle completed\n");
+                               syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
+                       IO = (AsyncIO *)chandle;
+                       
                        sta = msg->data.result;
                        if (sta) {
-                               CtdlLogPrintf(CTDL_ERR, "EVCURL: error description: %s\n", handle->errdesc);
-                               CtdlLogPrintf(CTDL_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
+                               syslog(LOG_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc);
+                               syslog(LOG_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
                        }
-                       long httpcode;
-                       sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &httpcode);
+                       sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode);
                        if (sta)
-                               CtdlLogPrintf(CTDL_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
-                       CtdlLogPrintf(CTDL_ERR, "EVCURL: http response code was %ld\n", (long)httpcode);
-                       CURLMcode msta = curl_multi_remove_handle(mhnd, chnd);
+                               syslog(LOG_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
+                       syslog(LOG_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode);
+                       msta = curl_multi_remove_handle(mhnd, chnd);
                        if (msta)
-                               CtdlLogPrintf(CTDL_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
-                       handle->attached = 0;
+                               syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
+
+                       IO->HttpReq.attached = 0;
+                       IO->SendDone(IO);
+                       curl_easy_cleanup(IO->HttpReq.chnd);
+                       curl_slist_free_all(IO->HttpReq.headers);
+                       FreeStrBuf(&IO->HttpReq.ReplyData);
+                       FreeURL(&IO->ConnectMe);
+                       RemoveContext(IO->CitContext);
+                       IO->Terminate(IO);
                }
        }
 }
@@ -146,38 +150,50 @@ gotstatus(evcurl_global_data *global, int nnrun)
 static void
 stepmulti(evcurl_global_data *global, curl_socket_t fd) {
        int nnrun;
-       CURLMcode msta = curl_multi_socket_action(global->mhnd, fd, 0, &nnrun);
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
+       CURLMcode msta;
+       
+       if (global == NULL) {
+           syslog(LOG_DEBUG, "EVCURL: stepmulti(NULL): wtf?\n");
+           return;
+       }
+       msta = curl_multi_socket_action(global->mhnd, fd, 0, &nnrun);
+       syslog(LOG_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
        if (msta)
-               CtdlLogPrintf(CTDL_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta));
+               syslog(LOG_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta));
        if (global->nrun != nnrun)
                gotstatus(global, nnrun);
 }
 
 static void
 gottime(struct ev_loop *loop, ev_timer *timeev, int events) {
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl for timeout\n");
+       syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n");
        evcurl_global_data *global = (void *)timeev->data;
        stepmulti(global, CURL_SOCKET_TIMEOUT);
 }
 
 static void
 gotio(struct ev_loop *loop, ev_io *ioev, int events) {
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
+       syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
        sockwatcher_data *sockwatcher = (void *)ioev->data;
        stepmulti(sockwatcher->global, ioev->fd);
 }
 
 static size_t
 gotdata(void *data, size_t size, size_t nmemb, void *cglobal) {
-       evcurl_request_data *D = (evcurl_request_data*) data;
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: gotdata(): calling CurlFillStrBuf_callback()\n");
-       return CurlFillStrBuf_callback(D->ReplyData, size, nmemb, cglobal);
+       AsyncIO *IO = (AsyncIO*) cglobal;
+       //evcurl_request_data *D = (evcurl_request_data*) data;
+       syslog(LOG_DEBUG, "EVCURL: gotdata(): calling CurlFillStrBuf_callback()\n");
+
+       if (IO->HttpReq.ReplyData == NULL)
+       {
+           IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
+       }
+       return CurlFillStrBuf_callback(data, size, nmemb, IO->HttpReq.ReplyData);
 }
 
 static int
 gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
+       syslog(LOG_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
        evcurl_global_data *global = cglobal;
        ev_timer_stop(EV_DEFAULT, &global->timeev);
        if (tblock_ms < 0 || 14000 < tblock_ms)
@@ -194,10 +210,10 @@ gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *csoc
        CURLM *mhnd = global->mhnd;
        sockwatcher_data *sockwatcher = csockwatcher;
 
-       CtdlLogPrintf(CTDL_DEBUG,"EVCURL: gotwatchsock called fd=%d action=%d\n", (int)fd, action);
+       syslog(LOG_DEBUG, "EVCURL: gotwatchsock called fd=%d action=%d\n", (int)fd, action);
 
        if (!sockwatcher) {
-               CtdlLogPrintf(CTDL_ERR,"EVCURL: called first time to register this sockwatcker\n");
+               syslog(LOG_ERR,"EVCURL: called first time to register this sockwatcker\n");
                sockwatcher = malloc(sizeof(sockwatcher_data));
                sockwatcher->global = global;
                ev_init(&sockwatcher->ioev, &gotio);
@@ -205,7 +221,8 @@ gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *csoc
                curl_multi_assign(mhnd, fd, sockwatcher);
        }
        if (CURL_POLL_REMOVE == action) {
-               CtdlLogPrintf(CTDL_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
+               syslog(LOG_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
+               ev_io_stop(event_base, &sockwatcher->ioev);
                free(sockwatcher);
        } else {
                int events = (action & CURL_POLL_IN ? EV_READ : 0) | (action & CURL_POLL_OUT ? EV_WRITE : 0);
@@ -221,7 +238,6 @@ gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *csoc
 void curl_init_connectionpool(void) 
 {
        CURLM *mhnd ;
-//     global.magic = EVCURL_GLOBAL_MAGIC;
 
        ev_timer_init(&global.timeev, &gottime, 14.0, 14.0);
        global.timeev.data = (void *)&global;
@@ -230,20 +246,13 @@ void curl_init_connectionpool(void)
 
        if (sta) 
        {
-               CtdlLogPrintf(CTDL_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta));
+               syslog(LOG_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta));
                exit(1);
        }
-/*
-  if (!ev_default_loop(EVFLAG_AUTO))
-  {
-  CtdlLogPrintf(CTDL_ERR,"error initializing libev\n");
-  exit(2);
-  }
-*/
        mhnd = global.mhnd = curl_multi_init();
        if (!mhnd)
        {
-               CtdlLogPrintf(CTDL_ERR,"EVCURL: error initializing curl multi handle\n");
+               syslog(LOG_ERR,"EVCURL: error initializing curl multi handle\n");
                exit(3);
        }
 
@@ -252,33 +261,33 @@ void curl_init_connectionpool(void)
        MOPT(TIMERFUNCTION, &gotwatchtime);
        MOPT(TIMERDATA, (void *)&global);
 
-       /* well, just there to fire the sample request?*/
-///    ev_timer_start(EV_DEFAULT, &global.timeev);
        return;
 }
 
 
 
 
-int evcurl_init(evcurl_request_data *handle
+int evcurl_init(AsyncIO *IO
                void *CustomData, 
                const char* Desc,
-               int CallBack) 
+               IO_CallBack CallBack, 
+               IO_CallBack Terminate)
 {
        CURLcode sta;
        CURL *chnd;
 
-       CtdlLogPrintf(CTDL_DEBUG,"EVCURL: evcurl_init called ms\n");
-       handle->global = &global;
-       handle->attached = 0;
-       chnd = handle->chnd = curl_easy_init();
+       syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
+       IO->HttpReq.attached = 0;
+       IO->SendDone = CallBack;
+       IO->Terminate = Terminate;
+       chnd = IO->HttpReq.chnd = curl_easy_init();
        if (!chnd)
        {
-               CtdlLogPrintf(CTDL_ERR, "EVCURL: error initializing curl handle\n");
+               syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
                return 1;
        }
 
-       strcpy(handle->errdesc, Desc);
+       strcpy(IO->HttpReq.errdesc, Desc);
 
        OPT(VERBOSE, (long)1);
                /* unset in production */
@@ -294,70 +303,91 @@ int evcurl_init(evcurl_request_data *handle,
        OPT(LOW_SPEED_LIMIT, (long)64);
        OPT(LOW_SPEED_TIME, (long)600);
        OPT(CONNECTTIMEOUT, (long)600); 
-       OPT(PRIVATE, (void *)handle);
+       OPT(PRIVATE, (void *)IO);
 
 
        OPT(WRITEFUNCTION, &gotdata); 
-       OPT(WRITEDATA, (void *)handle);
-       OPT(ERRORBUFFER, handle->errdesc);
-
+       OPT(WRITEDATA, (void *)IO);
+       OPT(ERRORBUFFER, IO->HttpReq.errdesc);
+
+       if (
+               (!IsEmptyStr(config.c_ip_addr))
+               && (strcmp(config.c_ip_addr, "*"))
+               && (strcmp(config.c_ip_addr, "::"))
+               && (strcmp(config.c_ip_addr, "0.0.0.0"))
+       ) {
+               OPT(INTERFACE, config.c_ip_addr);
+       }
                /* point to a structure that points back to the perl structure and stuff */
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: Loading URL: %s\n", handle->URL->PlainUrl);
-       OPT(URL, handle->URL->PlainUrl);
-       if (StrLength(handle->URL->CurlCreds))
+       syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
+       OPT(URL, IO->ConnectMe->PlainUrl);
+       if (StrLength(IO->ConnectMe->CurlCreds))
        {
                OPT(HTTPAUTH, (long)CURLAUTH_BASIC);
-               OPT(USERPWD, ChrPtr(handle->URL->CurlCreds));
+               OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds));
        }
 #ifdef CURLOPT_HTTP_CONTENT_DECODING
        OPT(HTTP_CONTENT_DECODING, 1);
        OPT(ENCODING, "");
 #endif
-       if (StrLength(handle->PostData) > 0)
+       if (StrLength(IO->HttpReq.PostData) > 0)
        { 
-               OPT(POSTFIELDS, ChrPtr(handle->PostData));
-               OPT(POSTFIELDSIZE, StrLength(handle->PostData));
+               OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData));
+               OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData));
 
        }
-       else if ((handle->PlainPostDataLen != 0) && (handle->PlainPostData != NULL))
+       else if ((IO->HttpReq.PlainPostDataLen != 0) && (IO->HttpReq.PlainPostData != NULL))
        {
-               OPT(POSTFIELDS, handle->PlainPostData);
-               OPT(POSTFIELDSIZE, handle->PlainPostDataLen);
+               OPT(POSTFIELDS, IO->HttpReq.PlainPostData);
+               OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen);
        }
 
-       if (handle->headers != NULL)
-               OPT(HTTPHEADER, handle->headers);
+       if (IO->HttpReq.headers != NULL)
+               OPT(HTTPHEADER, IO->HttpReq.headers);
 
        return 1;
 }
 
 void
-evcurl_handle_start(evcurl_request_data *handle
+evcurl_handle_start(AsyncIO *IO
 {
        CURLMcode msta;
        
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: attaching to curl multi handle\n");
-       msta = curl_multi_add_handle(handle->global->mhnd, handle->chnd);
+       syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
+       msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
        if (msta)
-               CtdlLogPrintf(CTDL_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
-       handle->attached = 1;
-//     ev_timer_start(EV_DEFAULT, &global.timeev);
+               syslog(LOG_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
+       IO->HttpReq.attached = 1;
        ev_async_send (event_base, &WakeupCurl);
 }
 
 static void WakeupCurlCallback(EV_P_ ev_async *w, int revents)
 {
-///    evcurl_global_data *global = cglobal;
-
-       CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl multi handle\n");
+       syslog(LOG_DEBUG, "EVCURL: waking up curl multi handle\n");
 
        curl_multi_perform(&global, CURL_POLL_NONE);
 }
 
+static void evcurl_shutdown (void)
+{
+       curl_multi_cleanup(global.mhnd);
+}
 /*****************************************************************************
  *                       libevent integration                                *
  *****************************************************************************/
+/*
+ * client event queue plus its methods.
+ * this currently is the main loop (which may change in some future?)
+ */
+int evbase_count = 0;
+int event_add_pipe[2] = {-1, -1};
+pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
+HashList *QueueEvents = NULL;
+HashList *InboundEventQueue = NULL;
+HashList *InboundEventQueues[2] = { NULL, NULL };
 
+ev_async AddJob;   
+ev_async ExitEventLoop;
 
 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 {
@@ -368,7 +398,7 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
        const char *Key;
 
        /* get the control command... */
-       citthread_mutex_lock(&EventQueueMutex);
+       pthread_mutex_lock(&EventQueueMutex);
 
        if (InboundEventQueues[0] == InboundEventQueue) {
                InboundEventQueue = InboundEventQueues[1];
@@ -378,7 +408,7 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
                InboundEventQueue = InboundEventQueues[0];
                q = InboundEventQueues[1];
        }
-       citthread_mutex_unlock(&EventQueueMutex);
+       pthread_mutex_unlock(&EventQueueMutex);
 
        It = GetNewHashPos(q, 0);
        while (GetNextHashPos(q, It, &len, &Key, &v))
@@ -388,15 +418,15 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
        }
        DeleteHashPos(&It);
        DeleteHashContent(&q);
-       CtdlLogPrintf(CTDL_DEBUG, "EVENT Q Read done.\n");
+       syslog(LOG_DEBUG, "EVENT Q Read done.\n");
 }
 
 
 static void EventExitCallback(EV_P_ ev_async *w, int revents)
 {
-       ev_unloop(event_base, EVUNLOOP_ALL);
+       ev_break(event_base, EVBREAK_ALL);
 
-       CtdlLogPrintf(CTDL_DEBUG, "EVENT Q exiting.\n");
+       syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 }
 
 
@@ -405,10 +435,10 @@ void InitEventQueue(void)
 {
        struct rlimit LimitSet;
 
-       citthread_mutex_init(&EventQueueMutex, NULL);
+       pthread_mutex_init(&EventQueueMutex, NULL);
 
        if (pipe(event_add_pipe) != 0) {
-               CtdlLogPrintf(CTDL_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
+               syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
                abort();
        }
        LimitSet.rlim_cur = 1;
@@ -427,11 +457,11 @@ void InitEventQueue(void)
  */
 void *client_event_thread(void *arg) 
 {
-       struct CitContext libevent_client_CC;
+       struct CitContext libev_client_CC;
 
-       CtdlFillSystemContext(&libevent_client_CC, "LibEv Thread");
+       CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
 //     citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
-       CtdlLogPrintf(CTDL_DEBUG, "client_ev_thread() initializing\n");
+       syslog(LOG_DEBUG, "client_ev_thread() initializing\n");
 
        event_base = ev_default_loop (EVFLAG_AUTO);
 
@@ -444,33 +474,152 @@ void *client_event_thread(void *arg)
 
        curl_init_connectionpool();
 
-       ev_loop (event_base, 0);
+       ev_run (event_base, 0);
 
 
-       CtdlClearSystemContext();
-       ev_default_destroy ();
+///what todo here?     CtdlClearSystemContext();
+       ev_loop_destroy (EV_DEFAULT_UC);
        
        DeleteHash(&QueueEvents);
        InboundEventQueue = NULL;
        DeleteHash(&InboundEventQueues[0]);
        DeleteHash(&InboundEventQueues[1]);
-       citthread_mutex_destroy(&EventQueueMutex);
-
+/*     citthread_mutex_destroy(&EventQueueMutex); TODO */
+       evcurl_shutdown();
 
        return(NULL);
 }
+/*------------------------------------------------------------------------------*/
+/*
+ * DB-Queue; does async bdb operations.
+ * has its own set of handlers.
+ */
+ev_loop *event_db;
+int evdb_count = 0;
+int evdb_add_pipe[2] = {-1, -1};
+pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */
+HashList *DBQueueEvents = NULL;
+HashList *DBInboundEventQueue = NULL;
+HashList *DBInboundEventQueues[2] = { NULL, NULL };
 
-#endif
+ev_async DBAddJob;   
+ev_async DBExitEventLoop;
+
+extern void ShutDownDBCLient(AsyncIO *IO);
+
+static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
+{
+       HashList *q;
+       void *v;
+       HashPos  *It;
+       long len;
+       const char *Key;
+
+       /* get the control command... */
+       pthread_mutex_lock(&DBEventQueueMutex);
+
+       if (DBInboundEventQueues[0] == DBInboundEventQueue) {
+               DBInboundEventQueue = DBInboundEventQueues[1];
+               q = DBInboundEventQueues[0];
+       }
+       else {
+               DBInboundEventQueue = DBInboundEventQueues[0];
+               q = DBInboundEventQueues[1];
+       }
+       pthread_mutex_unlock(&DBEventQueueMutex);
+
+       It = GetNewHashPos(q, 0);
+       while (GetNextHashPos(q, It, &len, &Key, &v))
+       {
+               IOAddHandler *h = v;
+               eNextState rc;
+               rc = h->EvAttch(h->IO);
+               switch (rc)
+               {
+               case eAbort:
+                   ShutDownDBCLient(h->IO);
+               default:
+                   break;
+               }
+       }
+       DeleteHashPos(&It);
+       DeleteHashContent(&q);
+       syslog(LOG_DEBUG, "DBEVENT Q Read done.\n");
+}
+
+
+static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
+{
+       syslog(LOG_DEBUG, "EVENT Q exiting.\n");
+       ev_break(event_db, EVBREAK_ALL);
+}
+
+
+
+void DBInitEventQueue(void)
+{
+       struct rlimit LimitSet;
+
+       pthread_mutex_init(&DBEventQueueMutex, NULL);
+
+       if (pipe(evdb_add_pipe) != 0) {
+               syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
+               abort();
+       }
+       LimitSet.rlim_cur = 1;
+       LimitSet.rlim_max = 1;
+       setrlimit(evdb_add_pipe[1], &LimitSet);
+
+       DBQueueEvents = NewHash(1, Flathash);
+       DBInboundEventQueues[0] = NewHash(1, Flathash);
+       DBInboundEventQueues[1] = NewHash(1, Flathash);
+       DBInboundEventQueue = DBInboundEventQueues[0];
+}
+
+/*
+ * this thread operates writing to the message database via libev.
+ * 
+ * 
+ */
+void *db_event_thread(void *arg) 
+{
+       struct CitContext libev_msg_CC;
+
+       CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread");
+//     citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
+       syslog(LOG_DEBUG, "client_msgev_thread() initializing\n");
+
+       event_db = ev_loop_new (EVFLAG_AUTO);
+
+       ev_async_init(&DBAddJob, DBQueueEventAddCallback);
+       ev_async_start(event_db, &DBAddJob);
+       ev_async_init(&DBExitEventLoop, DBEventExitCallback);
+       ev_async_start(event_db, &DBExitEventLoop);
+
+       ev_run (event_db, 0);
+
+
+//// what to do here?  CtdlClearSystemContext();
+       ev_loop_destroy (event_db);
+
+       DeleteHash(&DBQueueEvents);
+       DBInboundEventQueue = NULL;
+       DeleteHash(&DBInboundEventQueues[0]);
+       DeleteHash(&DBInboundEventQueues[1]);
+/*     citthread_mutex_destroy(&DBEventQueueMutex); TODO */
+
+       return(NULL);
+}
 
 CTDL_MODULE_INIT(event_client)
 {
-#ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT
        if (!threading)
        {
                InitEventQueue();
-               CtdlThreadCreate("Client event", CTDLTHREAD_BIGSTACK, client_event_thread, NULL);
+               DBInitEventQueue();
+               CtdlThreadCreate(/*"Client event", */ client_event_thread);
+               CtdlThreadCreate(/*"DB event", */db_event_thread);
 /// todo register shutdown callback.
        }
-#endif
        return "event";
 }