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 #ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT
  59
  60 #include "event_client.h"
  61 #include "serv_curl.h"
  62
  63 int event_add_pipe[2] = {-1, -1};
  64
  65 citthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */
  66 HashList *QueueEvents = NULL;
  67
  68 HashList *InboundEventQueue = NULL;
  69 HashList *InboundEventQueues[2] = { NULL, NULL };
  70
  71 struct ev_loop *event_base;
  72
  73 ev_async AddJob;
  74 ev_async ExitEventLoop;
  75 ev_async WakeupCurl;
  76
  77 extern struct ev_loop *event_base;
  78
  79 void SockStateCb(void *data, int sock, int read, int write);
  80
  81 #define MOPT(s, v)                                                      \
  82         do {                                                            \
  83                 sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v));     \
  84                 if (sta) {                                              \
  85                         CtdlLogPrintf(CTDL_ERR, "EVCURL: error setting option " #s " on curl multi handle: %s\n", curl_easy_strerror(sta)); \
  86                         exit (1);                                       \
  87                 }                                                       \
  88         } while (0)
  89
  90
  91
  92 /*****************************************************************************
  93  *                   libevent / curl integration                             *
  94  *****************************************************************************/
  95 typedef struct _evcurl_global_data {
  96         int magic;
  97         CURLM *mhnd;
  98         ev_timer timeev;
  99         int nrun;
 100 } evcurl_global_data;
 101
 102 typedef struct _sockwatcher_data
 103 {
 104         evcurl_global_data *global;
 105         ev_io ioev;
 106 } sockwatcher_data;
 107
 108 evcurl_global_data global;
 109
 110 static void
 111 gotstatus(evcurl_global_data *global, int nnrun)
 112 {
 113         CURLM *mhnd;
 114         CURLMsg *msg;
 115         int nmsg;
 116 /*
 117   if (EVCURL_GLOBAL_MAGIC != global.magic)
 118   {
 119   CtdlLogPrintf(CTDL_ERR, "internal error: gotstatus on wrong struct");
 120   return;
 121   }
 122 */
 123         global->nrun = nnrun;
 124         mhnd = global->mhnd;
 125
 126         CtdlLogPrintf(CTDL_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
 127         while ((msg = curl_multi_info_read(mhnd, &nmsg))) {
 128                 CtdlLogPrintf(CTDL_ERR, "EVCURL: got curl multi_info message msg=%d\n", msg->msg);
 129                 if (CURLMSG_DONE == msg->msg) {
 130                         CURL *chnd;
 131                         char *chandle;
 132                         CURLcode sta;
 133                         CURLMcode msta;
 134                         AsyncIO  *IO;
 135
 136                         chandle = NULL;;
 137                         chnd = msg->easy_handle;
 138                         sta = curl_easy_getinfo(chnd, CURLINFO_PRIVATE, &chandle);
 139                         CtdlLogPrintf(CTDL_ERR, "EVCURL: request complete\n");
 140                         if (sta)
 141                                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
 142                         IO = (AsyncIO *)chandle;
 143
 144                         sta = msg->data.result;
 145                         if (sta) {
 146                                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc);
 147                                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
 148                         }
 149                         sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode);
 150                         if (sta)
 151                                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
 152                         CtdlLogPrintf(CTDL_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode);
 153                         msta = curl_multi_remove_handle(mhnd, chnd);
 154                         if (msta)
 155                                 CtdlLogPrintf(CTDL_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 156                         IO->HttpReq.attached = 0;
 157                         IO->SendDone(IO);
 158                         curl_multi_cleanup(msta);
 159                 }
 160         }
 161 }
 162
 163 static void
 164 stepmulti(evcurl_global_data *global, curl_socket_t fd) {
 165         int nnrun;
 166         CURLMcode msta;
 167
 168         if (global == NULL) {
 169             CtdlLogPrintf(CTDL_DEBUG, "EVCURL: stepmulti(NULL): wtf?\n");
 170             return;
 171         }
 172         msta = curl_multi_socket_action(global->mhnd, fd, 0, &nnrun);
 173         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
 174         if (msta)
 175                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error in curl processing events on multi handle, fd %d: %s\n", (int)fd, curl_multi_strerror(msta));
 176         if (global->nrun != nnrun)
 177                 gotstatus(global, nnrun);
 178 }
 179
 180 static void
 181 gottime(struct ev_loop *loop, ev_timer *timeev, int events) {
 182         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl for timeout\n");
 183         evcurl_global_data *global = (void *)timeev->data;
 184         stepmulti(global, CURL_SOCKET_TIMEOUT);
 185 }
 186
 187 static void
 188 gotio(struct ev_loop *loop, ev_io *ioev, int events) {
 189         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
 190         sockwatcher_data *sockwatcher = (void *)ioev->data;
 191         stepmulti(sockwatcher->global, ioev->fd);
 192 }
 193
 194 static size_t
 195 gotdata(void *data, size_t size, size_t nmemb, void *cglobal) {
 196         AsyncIO *IO = (AsyncIO*) cglobal;
 197         //evcurl_request_data *D = (evcurl_request_data*) data;
 198         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: gotdata(): calling CurlFillStrBuf_callback()\n");
 199
 200         if (IO->HttpReq.ReplyData == NULL)
 201         {
 202             IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ);
 203         }
 204         return CurlFillStrBuf_callback(data, size, nmemb, IO->HttpReq.ReplyData);
 205 }
 206
 207 static int
 208 gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
 209         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: gotwatchtime called %ld ms\n", tblock_ms);
 210         evcurl_global_data *global = cglobal;
 211         ev_timer_stop(EV_DEFAULT, &global->timeev);
 212         if (tblock_ms < 0 || 14000 < tblock_ms)
 213                 tblock_ms = 14000;
 214         ev_timer_set(&global->timeev, 0.5e-3 + 1.0e-3 * tblock_ms, 14.0);
 215         ev_timer_start(EV_DEFAULT_UC, &global->timeev);
 216         curl_multi_perform(global, CURL_POLL_NONE);
 217         return 0;
 218 }
 219
 220 static int
 221 gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *csockwatcher) {
 222         evcurl_global_data *global = cglobal;
 223         CURLM *mhnd = global->mhnd;
 224         sockwatcher_data *sockwatcher = csockwatcher;
 225
 226         CtdlLogPrintf(CTDL_DEBUG,"EVCURL: gotwatchsock called fd=%d action=%d\n", (int)fd, action);
 227
 228         if (!sockwatcher) {
 229                 CtdlLogPrintf(CTDL_ERR,"EVCURL: called first time to register this sockwatcker\n");
 230                 sockwatcher = malloc(sizeof(sockwatcher_data));
 231                 sockwatcher->global = global;
 232                 ev_init(&sockwatcher->ioev, &gotio);
 233                 sockwatcher->ioev.data = (void *)sockwatcher;
 234                 curl_multi_assign(mhnd, fd, sockwatcher);
 235         }
 236         if (CURL_POLL_REMOVE == action) {
 237                 CtdlLogPrintf(CTDL_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
 238                 free(sockwatcher);
 239         } else {
 240                 int events = (action & CURL_POLL_IN ? EV_READ : 0) | (action & CURL_POLL_OUT ? EV_WRITE : 0);
 241                 ev_io_stop(EV_DEFAULT, &sockwatcher->ioev);
 242                 if (events) {
 243                         ev_io_set(&sockwatcher->ioev, fd, events);
 244                         ev_io_start(EV_DEFAULT, &sockwatcher->ioev);
 245                 }
 246         }
 247         return 0;
 248 }
 249
 250 void curl_init_connectionpool(void)
 251 {
 252         CURLM *mhnd ;
 253 //      global.magic = EVCURL_GLOBAL_MAGIC;
 254
 255         ev_timer_init(&global.timeev, &gottime, 14.0, 14.0);
 256         global.timeev.data = (void *)&global;
 257         global.nrun = -1;
 258         CURLcode sta = curl_global_init(CURL_GLOBAL_ALL);
 259
 260         if (sta)
 261         {
 262                 CtdlLogPrintf(CTDL_ERR,"EVCURL: error initializing curl library: %s\n", curl_easy_strerror(sta));
 263                 exit(1);
 264         }
 265 /*
 266   if (!ev_default_loop(EVFLAG_AUTO))
 267   {
 268   CtdlLogPrintf(CTDL_ERR,"error initializing libev\n");
 269   exit(2);
 270   }
 271 */
 272         mhnd = global.mhnd = curl_multi_init();
 273         if (!mhnd)
 274         {
 275                 CtdlLogPrintf(CTDL_ERR,"EVCURL: error initializing curl multi handle\n");
 276                 exit(3);
 277         }
 278
 279         MOPT(SOCKETFUNCTION, &gotwatchsock);
 280         MOPT(SOCKETDATA, (void *)&global);
 281         MOPT(TIMERFUNCTION, &gotwatchtime);
 282         MOPT(TIMERDATA, (void *)&global);
 283
 284         /* well, just there to fire the sample request?*/
 285 ///     ev_timer_start(EV_DEFAULT, &global.timeev);
 286         return;
 287 }
 288
 289
 290
 291
 292 int evcurl_init(AsyncIO *IO,
 293                 void *CustomData,
 294                 const char* Desc,
 295                 IO_CallBack CallBack)
 296 {
 297         CURLcode sta;
 298         CURL *chnd;
 299
 300         CtdlLogPrintf(CTDL_DEBUG,"EVCURL: evcurl_init called ms\n");
 301         IO->HttpReq.attached = 0;
 302         IO->SendDone = CallBack;
 303         chnd = IO->HttpReq.chnd = curl_easy_init();
 304         if (!chnd)
 305         {
 306                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error initializing curl handle\n");
 307                 return 1;
 308         }
 309
 310         strcpy(IO->HttpReq.errdesc, Desc);
 311
 312         OPT(VERBOSE, (long)1);
 313                 /* unset in production */
 314         OPT(NOPROGRESS, (long)1);
 315         OPT(NOSIGNAL, (long)1);
 316         OPT(FAILONERROR, (long)1);
 317         OPT(ENCODING, "");
 318         OPT(FOLLOWLOCATION, (long)1);
 319         OPT(MAXREDIRS, (long)7);
 320         OPT(USERAGENT, CITADEL);
 321
 322         OPT(TIMEOUT, (long)1800);
 323         OPT(LOW_SPEED_LIMIT, (long)64);
 324         OPT(LOW_SPEED_TIME, (long)600);
 325         OPT(CONNECTTIMEOUT, (long)600);
 326         OPT(PRIVATE, (void *)IO);
 327
 328
 329         OPT(WRITEFUNCTION, &gotdata);
 330         OPT(WRITEDATA, (void *)IO);
 331         OPT(ERRORBUFFER, IO->HttpReq.errdesc);
 332
 333                 /* point to a structure that points back to the perl structure and stuff */
 334         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
 335         OPT(URL, IO->ConnectMe->PlainUrl);
 336         if (StrLength(IO->ConnectMe->CurlCreds))
 337         {
 338                 OPT(HTTPAUTH, (long)CURLAUTH_BASIC);
 339                 OPT(USERPWD, ChrPtr(IO->ConnectMe->CurlCreds));
 340         }
 341 #ifdef CURLOPT_HTTP_CONTENT_DECODING
 342         OPT(HTTP_CONTENT_DECODING, 1);
 343         OPT(ENCODING, "");
 344 #endif
 345         if (StrLength(IO->HttpReq.PostData) > 0)
 346         {
 347                 OPT(POSTFIELDS, ChrPtr(IO->HttpReq.PostData));
 348                 OPT(POSTFIELDSIZE, StrLength(IO->HttpReq.PostData));
 349
 350         }
 351         else if ((IO->HttpReq.PlainPostDataLen != 0) && (IO->HttpReq.PlainPostData != NULL))
 352         {
 353                 OPT(POSTFIELDS, IO->HttpReq.PlainPostData);
 354                 OPT(POSTFIELDSIZE, IO->HttpReq.PlainPostDataLen);
 355         }
 356
 357         if (IO->HttpReq.headers != NULL)
 358                 OPT(HTTPHEADER, IO->HttpReq.headers);
 359
 360         return 1;
 361 }
 362
 363 void
 364 evcurl_handle_start(AsyncIO *IO)
 365 {
 366         CURLMcode msta;
 367
 368         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: attaching to curl multi handle\n");
 369         msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
 370         if (msta)
 371                 CtdlLogPrintf(CTDL_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
 372         IO->HttpReq.attached = 1;
 373 //      ev_timer_start(EV_DEFAULT, &global.timeev);
 374         ev_async_send (event_base, &WakeupCurl);
 375 }
 376
 377 static void WakeupCurlCallback(EV_P_ ev_async *w, int revents)
 378 {
 379 ///     evcurl_global_data *global = cglobal;
 380
 381         CtdlLogPrintf(CTDL_DEBUG, "EVCURL: waking up curl multi handle\n");
 382
 383         curl_multi_perform(&global, CURL_POLL_NONE);
 384 }
 385
 386 /*****************************************************************************
 387  *                       libevent integration                                *
 388  *****************************************************************************/
 389
 390
 391 static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
 392 {
 393         HashList *q;
 394         void *v;
 395         HashPos  *It;
 396         long len;
 397         const char *Key;
 398
 399         /* get the control command... */
 400         citthread_mutex_lock(&EventQueueMutex);
 401
 402         if (InboundEventQueues[0] == InboundEventQueue) {
 403                 InboundEventQueue = InboundEventQueues[1];
 404                 q = InboundEventQueues[0];
 405         }
 406         else {
 407                 InboundEventQueue = InboundEventQueues[0];
 408                 q = InboundEventQueues[1];
 409         }
 410         citthread_mutex_unlock(&EventQueueMutex);
 411
 412         It = GetNewHashPos(q, 0);
 413         while (GetNextHashPos(q, It, &len, &Key, &v))
 414         {
 415                 IOAddHandler *h = v;
 416                 h->EvAttch(h->IO);
 417         }
 418         DeleteHashPos(&It);
 419         DeleteHashContent(&q);
 420         CtdlLogPrintf(CTDL_DEBUG, "EVENT Q Read done.\n");
 421 }
 422
 423
 424 static void EventExitCallback(EV_P_ ev_async *w, int revents)
 425 {
 426         ev_unloop(event_base, EVUNLOOP_ALL);
 427
 428         CtdlLogPrintf(CTDL_DEBUG, "EVENT Q exiting.\n");
 429 }
 430
 431
 432
 433 void InitEventQueue(void)
 434 {
 435         struct rlimit LimitSet;
 436
 437         citthread_mutex_init(&EventQueueMutex, NULL);
 438
 439         if (pipe(event_add_pipe) != 0) {
 440                 CtdlLogPrintf(CTDL_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno));
 441                 abort();
 442         }
 443         LimitSet.rlim_cur = 1;
 444         LimitSet.rlim_max = 1;
 445         setrlimit(event_add_pipe[1], &LimitSet);
 446
 447         QueueEvents = NewHash(1, Flathash);
 448         InboundEventQueues[0] = NewHash(1, Flathash);
 449         InboundEventQueues[1] = NewHash(1, Flathash);
 450         InboundEventQueue = InboundEventQueues[0];
 451 }
 452 /*
 453  * this thread operates the select() etc. via libev.
 454  *
 455  *
 456  */
 457 void *client_event_thread(void *arg)
 458 {
 459         struct CitContext libevent_client_CC;
 460
 461         CtdlFillSystemContext(&libevent_client_CC, "LibEv Thread");
 462 //      citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 463         CtdlLogPrintf(CTDL_DEBUG, "client_ev_thread() initializing\n");
 464
 465         event_base = ev_default_loop (EVFLAG_AUTO);
 466
 467         ev_async_init(&AddJob, QueueEventAddCallback);
 468         ev_async_start(event_base, &AddJob);
 469         ev_async_init(&ExitEventLoop, EventExitCallback);
 470         ev_async_start(event_base, &ExitEventLoop);
 471         ev_async_init(&WakeupCurl, WakeupCurlCallback);
 472         ev_async_start(event_base, &WakeupCurl);
 473
 474         curl_init_connectionpool();
 475
 476         ev_loop (event_base, 0);
 477
 478
 479         CtdlClearSystemContext();
 480         ev_default_destroy ();
 481
 482         DeleteHash(&QueueEvents);
 483         InboundEventQueue = NULL;
 484         DeleteHash(&InboundEventQueues[0]);
 485         DeleteHash(&InboundEventQueues[1]);
 486         citthread_mutex_destroy(&EventQueueMutex);
 487
 488
 489         return(NULL);
 490 }
 491
 492 #endif
 493
 494 CTDL_MODULE_INIT(event_client)
 495 {
 496 #ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT
 497         if (!threading)
 498         {
 499                 InitEventQueue();
 500                 CtdlThreadCreate("Client event", CTDLTHREAD_BIGSTACK, client_event_thread, NULL);
 501 /// todo register shutdown callback.
 502         }
 503 #endif
 504         return "event";
 505 }