]> code.citadel.org Git - citadel.git/blobdiff - citadel/modules/eventclient/serv_eventclient.c
Work on evented RSS client & libev+libcurl integration
[citadel.git] / citadel / modules / eventclient / serv_eventclient.c
index 96201d655937166ac11f08493b571ed2a1c12f9d..153f1847ec1dcd79956bf67b03a8ce6a0d4a481c 100644 (file)
@@ -60,6 +60,7 @@
 
 ev_loop *event_base;
 
+long EvIDSource = 1;
 /*****************************************************************************
  *                   libevent / curl integration                             *
  *****************************************************************************/
@@ -83,14 +84,14 @@ ev_async WakeupCurl;
 evcurl_global_data global;
 
 static void
-gotstatus(evcurl_global_data *global, int nnrun) 
+gotstatus(int nnrun) 
 {
         CURLM *mhnd;
         CURLMsg *msg;
         int nmsg;
 
-        global->nrun = nnrun;
-        mhnd = global->mhnd;
+        global.nrun = nnrun;
+        mhnd = global.mhnd;
 
         syslog(LOG_DEBUG, "CURLEV: gotstatus(): about to call curl_multi_info_read\n");
         while ((msg = curl_multi_info_read(mhnd, &nmsg))) {
@@ -110,75 +111,90 @@ gotstatus(evcurl_global_data *global, int nnrun)
                                 syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
                         IO = (AsyncIO *)chandle;
                         
-/////                        ev_io_stop(event_base, &IO->recv_event);
+                        ev_io_stop(event_base, &IO->recv_event);
+                        ev_io_stop(event_base, &IO->send_event);
 
                         sta = msg->data.result;
                         if (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));
+                                EV_syslog(LOG_ERR, "EVCURL: error description: %s\n", IO->HttpReq.errdesc);
+                                EV_syslog(LOG_ERR, "EVCURL: error performing request: %s\n", curl_easy_strerror(sta));
                         }
                         sta = curl_easy_getinfo(chnd, CURLINFO_RESPONSE_CODE, &IO->HttpReq.httpcode);
                         if (sta)
-                                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);
-///                     syslog(LOG_ERR, "EVCURL: disconnecting [%s]\nn", ChrPtr(IO->ConnectMe->URL));
-                        msta = curl_multi_remove_handle(mhnd, chnd);
-                        if (msta)
-                                syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
-///                     syslog(LOG_ERR, "EVCURL: disconnection done [%s]\nn", ChrPtr(IO->ConnectMe->URL));
+                                EV_syslog(LOG_ERR, "EVCURL: error asking curl for response code from request: %s\n", curl_easy_strerror(sta));
+                        EV_syslog(LOG_ERR, "EVCURL: http response code was %ld\n", (long)IO->HttpReq.httpcode);
 
-                        IO->HttpReq.attached = 0;
-                        IO->SendDone(IO);
-////                    syslog(LOG_ERR, "EVCURL: done with [%s]\nn", ChrPtr(IO->ConnectMe->URL));
 
+                        curl_slist_free_all(IO->HttpReq.headers);
+                        msta = curl_multi_remove_handle(mhnd, chnd);
+                        if (msta)
+                                EV_syslog(LOG_ERR, "EVCURL: warning problem detaching completed handle from curl multi: %s\n", curl_multi_strerror(msta));
 
                         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);
-*/
+                       IO->HttpReq.chnd = NULL;
+
+                        IO->HttpReq.attached = 0;
+                        switch(IO->SendDone(IO))
+                       {
+                       case eDBQuery:
+                               break;
+                       case eSendDNSQuery:
+                       case eReadDNSReply:
+                       case eConnect:
+                       case eSendReply: 
+                       case eSendMore:
+                       case eSendFile:
+                       case eReadMessage: 
+                       case eReadMore:
+                       case eReadPayload:
+                       case eReadFile:
+                               break;
+                       case eTerminateConnection:
+                       case eAbort:
+                               FreeStrBuf(&IO->HttpReq.ReplyData);
+                               FreeURL(&IO->ConnectMe);
+                               RemoveContext(IO->CitContext);
+                               IO->Terminate(IO);
+                       }
                 }
         }
 }
 
 static void
-stepmulti(evcurl_global_data *global, curl_socket_t fd) {
-        int nnrun;
+stepmulti(void *data, curl_socket_t fd, int which)
+{
+        int running_handles = 0;
         CURLMcode msta;
         
-        if (global == NULL) {
-            syslog(LOG_DEBUG, "EVCURL: stepmulti(NULL): wtf?\n");
-            return;
-        }
-        msta = curl_multi_socket_action(global->mhnd, fd, 0, &nnrun);
+        msta = curl_multi_socket_action(global.mhnd, fd, which, &running_handles);
         syslog(LOG_DEBUG, "EVCURL: stepmulti(): calling gotstatus()\n");
         if (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);
+        if (global.nrun != running_handles)
+                gotstatus(running_handles);
 }
 
 static void
 gottime(struct ev_loop *loop, ev_timer *timeev, int events) {
         syslog(LOG_DEBUG, "EVCURL: waking up curl for timeout\n");
-        evcurl_global_data *global = (void *)timeev->data;
-        stepmulti(global, CURL_SOCKET_TIMEOUT);
+        stepmulti(NULL, CURL_SOCKET_TIMEOUT, 0);
+}
+
+static void
+got_in(struct ev_loop *loop, ev_io *ioev, int events) {
+        syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
+        stepmulti(ioev->data, ioev->fd, CURL_CSELECT_IN);
 }
 
 static void
-gotio(struct ev_loop *loop, ev_io *ioev, int events) {
+got_out(struct ev_loop *loop, ev_io *ioev, int events) {
         syslog(LOG_DEBUG, "EVCURL: waking up curl for io on fd %d\n", (int)ioev->fd);
-        stepmulti(&global, ioev->fd);
+        stepmulti(ioev->data, ioev->fd, CURL_CSELECT_OUT);
 }
 
 static size_t
 gotdata(void *data, size_t size, size_t nmemb, void *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)
         {
@@ -201,40 +217,78 @@ gotwatchtime(CURLM *multi, long tblock_ms, void *cglobal) {
 }
 
 static int
-gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *vio) {
+gotwatchsock(CURL *easy, curl_socket_t fd, int action, void *cglobal, void *vIO) {
         evcurl_global_data *global = cglobal;
         CURLM *mhnd = global->mhnd;
         char *f;
-        AsyncIO *IO = (AsyncIO*) vio;
+        AsyncIO *IO = (AsyncIO*) vIO;
         CURLcode sta;
+       const char *Action;
 
-        syslog(LOG_DEBUG, "EVCURL: gotwatchsock called fd=%d action=%d\n", (int)fd, action);
-
-        if (IO == NULL) {
-                syslog(LOG_ERR,"EVCURL: called first time to register this sockwatcker\n");
+       if (IO == NULL) {
                 sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f);
                 if (sta) {
-                        syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
+                        EV_syslog(LOG_ERR, "EVCURL: error asking curl for private cookie of curl handle: %s\n", curl_easy_strerror(sta));
                         return -1;
                 }
                 IO = (AsyncIO *) f;
-                ev_init(&IO->recv_event, &gotio);
-                curl_multi_assign(mhnd, fd, IO);
+               EV_syslog(LOG_DEBUG, "EVCURL: got socket for URL: %s\n", IO->ConnectMe->PlainUrl);
+               if (IO->SendBuf.fd != 0)
+               {
+                       ev_io_stop(event_base, &IO->recv_event);
+                       ev_io_stop(event_base, &IO->send_event);
+               }
+               IO->SendBuf.fd = fd;
+               ev_io_init(&IO->recv_event, &got_in, fd, EV_READ);
+               ev_io_init(&IO->send_event, &got_out, fd, EV_WRITE);
+               curl_multi_assign(mhnd, fd, IO);
+       }
+
+       Action = "";
+       switch (action)
+       {
+       case CURL_POLL_NONE:
+                Action = "CURL_POLL_NONE";
+               break;
+       case CURL_POLL_REMOVE:
+                Action = "CURL_POLL_REMOVE";
+               break;
+       case CURL_POLL_IN:
+                Action = "CURL_POLL_IN";
+               break;
+       case CURL_POLL_OUT:
+                Action = "CURL_POLL_OUT";
+               break;
+       case CURL_POLL_INOUT:
+                Action = "CURL_POLL_INOUT";
+               break;
         }
-        if (action == CURL_POLL_REMOVE) {
-                syslog(LOG_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
+
+
+        EV_syslog(LOG_DEBUG, "EVCURL: gotwatchsock called fd=%d action=%s[%d]\n", (int)fd, Action, action);
+
+       switch (action)
+       {
+       case CURL_POLL_NONE:
+                EVM_syslog(LOG_ERR,"EVCURL: called first time to register this sockwatcker\n");
+               break;
+       case CURL_POLL_REMOVE:
+                EVM_syslog(LOG_ERR,"EVCURL: called last time to unregister this sockwatcher\n");
                 ev_io_stop(event_base, &IO->recv_event);
-                FreeStrBuf(&IO->HttpReq.ReplyData);
-                FreeURL(&IO->ConnectMe);
-                RemoveContext(IO->CitContext);
-                IO->Terminate(IO);
-        } else {
-                int events = (action & CURL_POLL_IN ? EV_READ : 0) | (action & CURL_POLL_OUT ? EV_WRITE : 0);
+                ev_io_stop(event_base, &IO->send_event);
+               break;
+       case CURL_POLL_IN:
+                ev_io_start(event_base, &IO->recv_event);
+                ev_io_stop(event_base, &IO->send_event);
+               break;
+       case CURL_POLL_OUT:
+                ev_io_start(event_base, &IO->send_event);
                 ev_io_stop(event_base, &IO->recv_event);
-                if (events) {
-                        ev_io_set(&IO->recv_event, fd, events);
-                        ev_io_start(event_base, &IO->recv_event);
-                }
+               break;
+       case CURL_POLL_INOUT:
+                ev_io_start(event_base, &IO->send_event);
+                ev_io_start(event_base, &IO->recv_event);
+               break;
         }
         return 0;
 }
@@ -280,14 +334,14 @@ int evcurl_init(AsyncIO *IO,
         CURLcode sta;
         CURL *chnd;
 
-        syslog(LOG_DEBUG, "EVCURL: evcurl_init called ms\n");
+        EVM_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)
         {
-                syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
+                EVM_syslog(LOG_ERR, "EVCURL: error initializing curl handle\n");
                 return 1;
         }
 
@@ -299,7 +353,7 @@ int evcurl_init(AsyncIO *IO,
         OPT(NOSIGNAL, (long)1);
         OPT(FAILONERROR, (long)1);
         OPT(ENCODING, "");
-        OPT(FOLLOWLOCATION, (long)1);
+        OPT(FOLLOWLOCATION, (long)0);
         OPT(MAXREDIRS, (long)7);
         OPT(USERAGENT, CITADEL);
 
@@ -323,7 +377,7 @@ int evcurl_init(AsyncIO *IO,
                OPT(INTERFACE, config.c_ip_addr);
        }
                /* point to a structure that points back to the perl structure and stuff */
-       syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
+       EV_syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl);
        OPT(URL, IO->ConnectMe->PlainUrl);
        if (StrLength(IO->ConnectMe->CurlCreds))
        {
@@ -356,11 +410,11 @@ eNextState
 evcurl_handle_start(AsyncIO *IO) 
 {
        CURLMcode msta;
-       
-       syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
+       IO->NextState = eConnect;
+       EVM_syslog(LOG_DEBUG, "EVCURL: attaching to curl multi handle\n");
        msta = curl_multi_add_handle(global.mhnd, IO->HttpReq.chnd);
        if (msta)
-               syslog(LOG_ERR, "EVCURL: error attaching to curl multi handle: %s\n", curl_multi_strerror(msta));
+               EV_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);
        return eReadMessage;
@@ -419,11 +473,13 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents)
        while (GetNextHashPos(q, It, &len, &Key, &v))
        {
                IOAddHandler *h = v;
+               if (h->IO->ID == 0)
+                       h->IO->ID = EvIDSource++;
                h->EvAttch(h->IO);
        }
        DeleteHashPos(&It);
        DeleteHashContent(&q);
-       syslog(LOG_DEBUG, "EVENT Q Read done.\n");
+       syslog(LOG_DEBUG, "EVENT Q Add done.\n");
 }
 
 
@@ -538,6 +594,8 @@ static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
        {
                IOAddHandler *h = v;
                eNextState rc;
+               if (h->IO->ID == 0)
+                       h->IO->ID = EvIDSource++;
                rc = h->EvAttch(h->IO);
                switch (rc)
                {
@@ -549,7 +607,7 @@ static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents)
        }
        DeleteHashPos(&It);
        DeleteHashContent(&q);
-       syslog(LOG_DEBUG, "DBEVENT Q Read done.\n");
+       syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
 }