X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Feventclient%2Fserv_eventclient.c;h=06af6147647879ab460027923fa42908e5f726b6;hb=aaa8a12d698b952df5fc3499a13e423119db2447;hp=7b6a8c3dd27e8ca77ea6d3c668a617128c572b2e;hpb=2f6ccb7ba59f1aa4de0170ae2c8687cbfad15ce0;p=citadel.git

diff --git a/citadel/modules/eventclient/serv_eventclient.c b/citadel/modules/eventclient/serv_eventclient.c
index 7b6a8c3dd..06af61476 100644
--- a/citadel/modules/eventclient/serv_eventclient.c
+++ b/citadel/modules/eventclient/serv_eventclient.c
@@ -55,45 +55,37 @@
 
 #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";
 }