X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Frssclient%2Fserv_rssclient.c;h=82e206792bb77ea32d404e609769be0911341929;hb=222d2f63b0f523ce6b79e4b2ca9767f80e0bf85e;hp=0fda7a4577c53ca362d36b3d3cc858212ad7f705;hpb=99388bda231aa067339ca93539d3d3bb14fedcb1;p=citadel.git diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c index 0fda7a457..82e206792 100644 --- a/citadel/modules/rssclient/serv_rssclient.c +++ b/citadel/modules/rssclient/serv_rssclient.c @@ -1,8 +1,15 @@ /* - * $Id$ - * * Bring external RSS feeds into rooms. * + * Copyright (c) 2007-2012 by the citadel.org team + * + * This program is open source software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.See the + * GNU General Public License for more details. */ #include @@ -14,9 +21,9 @@ # include #else # if HAVE_SYS_TIME_H -# include +#include # else -# include +#include # endif #endif @@ -34,564 +41,772 @@ #include "support.h" #include "config.h" #include "threads.h" -#include "room_ops.h" #include "ctdl_module.h" -#include "clientsocket.h" #include "msgbase.h" #include "parsedate.h" #include "database.h" #include "citadel_dirs.h" #include "md5.h" +#include "context.h" +#include "event_client.h" +#include "rss_atom_parser.h" -struct rssnetcfg { - struct rssnetcfg *next; - char url[256]; - char *rooms; -}; - -struct rss_item { - char *chardata; - int chardata_len; - char *roomlist; - int done_parsing; - char *guid; - char *title; - char *link; - char *description; - time_t pubdate; - char channel_title[256]; - int item_tag_nesting; - char *author_or_creator; -}; +#define TMP_MSGDATA 0xFF +#define TMP_SHORTER_URL_OFFSET 0xFE +#define TMP_SHORTER_URLS 0xFD -struct rssnetcfg *rnclist = NULL; +time_t last_run = 0L; +pthread_mutex_t RSSQueueMutex; /* locks the access to the following vars: */ +HashList *RSSQueueRooms = NULL; /* rss_room_counter */ +HashList *RSSFetchUrls = NULL; /*->rss_aggregator;->RefCount access locked*/ -/* - * Commit a fetched and parsed RSS item to disk - */ -void rss_save_item(struct rss_item *ri) { +eNextState RSSAggregator_Terminate(AsyncIO *IO); +eNextState RSSAggregator_TerminateDB(AsyncIO *IO); +eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO); +struct CitContext rss_CC; - struct MD5Context md5context; - u_char rawdigest[MD5_DIGEST_LEN]; - int i; - char utmsgid[SIZ]; - struct cdbdata *cdbut; - struct UseTable ut; - struct CtdlMessage *msg; - struct recptypes *recp = NULL; - int msglen = 0; +struct rssnetcfg *rnclist = NULL; +int RSSClientDebugEnabled = 0; +#define N ((rss_aggregator*)IO->Data)->QRnumber - recp = (struct recptypes *) malloc(sizeof(struct recptypes)); - if (recp == NULL) return; - memset(recp, 0, sizeof(struct recptypes)); - recp->recp_room = strdup(ri->roomlist); - recp->num_room = num_tokens(ri->roomlist, '|'); - recp->recptypes_magic = RECPTYPES_MAGIC; - - /* Construct a GUID to use in the S_USETABLE table. - * If one is not present in the item itself, make one up. - */ - if (ri->guid != NULL) { - snprintf(utmsgid, sizeof utmsgid, "rss/%s", ri->guid); - } - else { - MD5Init(&md5context); - if (ri->title != NULL) { - MD5Update(&md5context, (unsigned char*)ri->title, strlen(ri->title)); - } - if (ri->link != NULL) { - MD5Update(&md5context, (unsigned char*)ri->link, strlen(ri->link)); - } - MD5Final(rawdigest, &md5context); - for (i=0; iID, CCID, N, __VA_ARGS__) - /* rewrite the record anyway, to update the timestamp */ - strcpy(ut.ut_msgid, utmsgid); - ut.ut_timestamp = time(NULL); - cdb_store(CDB_USETABLE, utmsgid, strlen(utmsgid), &ut, sizeof(struct UseTable) ); - } - else { - /* Item has not been seen, so save it. */ +#define EVRSSCM_syslog(LEVEL, FORMAT) \ + DBGLOG(LEVEL) syslog(LEVEL, \ + "IO[%ld]CC[%d][%ld]RSS" FORMAT, \ + IO->ID, CCID, N) - if (ri->description == NULL) ri->description = strdup(""); - for (i=strlen(ri->description); i>=0; --i) { - if (isspace(ri->description[i])) { - ri->description[i] = ' '; - } - } +#define EVRSSQ_syslog(LEVEL, FORMAT, ...) \ + DBGLOG(LEVEL) syslog(LEVEL, "RSS" FORMAT, \ + __VA_ARGS__) +#define EVRSSQM_syslog(LEVEL, FORMAT) \ + DBGLOG(LEVEL) syslog(LEVEL, "RSS" FORMAT) - msg = malloc(sizeof(struct CtdlMessage)); - memset(msg, 0, sizeof(struct CtdlMessage)); - msg->cm_magic = CTDLMESSAGE_MAGIC; - msg->cm_anon_type = MES_NORMAL; - msg->cm_format_type = FMT_RFC822; +#define EVRSSCSM_syslog(LEVEL, FORMAT) \ + DBGLOG(LEVEL) syslog(LEVEL, "IO[%ld][%ld]RSS" FORMAT, \ + IO->ID, N) - if (ri->author_or_creator != NULL) { - msg->cm_fields['A'] = html_to_ascii(ri->author_or_creator, - strlen(ri->author_or_creator), 512, 0); - striplt(msg->cm_fields['A']); - } - else { - msg->cm_fields['A'] = strdup("rss"); - } +void DeleteRoomReference(long QRnumber) +{ + HashPos *At; + long HKLen; + const char *HK; + void *vData = NULL; + rss_room_counter *pRoomC; + + At = GetNewHashPos(RSSQueueRooms, 0); - msg->cm_fields['N'] = strdup(NODENAME); - if (ri->title != NULL) { - msg->cm_fields['U'] = html_to_ascii(ri->title, strlen(ri->title), 512, 0); - striplt(msg->cm_fields['U']); + if (GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At)) + { + GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData); + if (vData != NULL) + { + pRoomC = (rss_room_counter *) vData; + pRoomC->count --; + if (pRoomC->count == 0) + DeleteEntryFromHash(RSSQueueRooms, At); } - msg->cm_fields['T'] = malloc(64); - snprintf(msg->cm_fields['T'], 64, "%ld", ri->pubdate); - if (ri->channel_title != NULL) { - if (!IsEmptyStr(ri->channel_title)) { - msg->cm_fields['O'] = strdup(ri->channel_title); - } + } + DeleteHashPos(&At); +} + +void UnlinkRooms(rss_aggregator *RSSAggr) +{ + DeleteRoomReference(RSSAggr->QRnumber); + if (RSSAggr->OtherQRnumbers != NULL) + { + long HKLen; + const char *HK; + HashPos *At; + void *vData; + + At = GetNewHashPos(RSSAggr->OtherQRnumbers, 0); + while (! server_shutting_down && + GetNextHashPos(RSSAggr->OtherQRnumbers, + At, + &HKLen, &HK, + &vData) && + (vData != NULL)) + { + long *lData = (long*) vData; + DeleteRoomReference(*lData); } - msglen = 1024 + strlen(ri->link) + strlen(ri->description) ; - msg->cm_fields['M'] = malloc(msglen); - snprintf(msg->cm_fields['M'], msglen, - "Content-type: text/html\r\n\r\n" - "\n" - "%s

\n" - "%s\n" - "\n" - , - ri->description, - ri->link, ri->link - ); + DeleteHashPos(&At); + } +} - CtdlSubmitMsg(msg, recp, NULL, 0); - CtdlFreeMessage(msg); +void UnlinkRSSAggregator(rss_aggregator *RSSAggr) +{ + HashPos *At; + + pthread_mutex_lock(&RSSQueueMutex); + UnlinkRooms(RSSAggr); - /* write the uidl to the use table so we don't store this item again */ - strcpy(ut.ut_msgid, utmsgid); - ut.ut_timestamp = time(NULL); - cdb_store(CDB_USETABLE, utmsgid, strlen(utmsgid), &ut, sizeof(struct UseTable) ); + At = GetNewHashPos(RSSFetchUrls, 0); + if (GetHashPosFromKey(RSSFetchUrls, SKEY(RSSAggr->Url), At)) + { + DeleteEntryFromHash(RSSFetchUrls, At); } - free_recipients(recp); + DeleteHashPos(&At); + last_run = time(NULL); + pthread_mutex_unlock(&RSSQueueMutex); } +void DeleteRssCfg(void *vptr) +{ + rss_aggregator *RSSAggr = (rss_aggregator *)vptr; + AsyncIO *IO = &RSSAggr->IO; + EVRSSCM_syslog(LOG_DEBUG, "RSS: destroying\n"); + FreeStrBuf(&RSSAggr->Url); + FreeStrBuf(&RSSAggr->rooms); + FreeStrBuf(&RSSAggr->CData); + FreeStrBuf(&RSSAggr->Key); + DeleteHash(&RSSAggr->OtherQRnumbers); -/* - * Convert an RDF/RSS datestamp into a time_t - */ -time_t rdf_parsedate(char *p) -{ - struct tm tm; - time_t t = 0; + DeleteHashPos (&RSSAggr->Pos); + DeleteHash (&RSSAggr->Messages); + if (RSSAggr->recp.recp_room != NULL) + free(RSSAggr->recp.recp_room); - if (!p) return 0L; - if (strlen(p) < 10) return 0L; - memset(&tm, 0, sizeof tm); + if (RSSAggr->Item != NULL) + { + flush_rss_item(RSSAggr->Item); - /* YYYY-MM-DDTHH:MM format... - */ - if ( (p[4] == '-') && (p[7] == '-') ) { - tm.tm_year = atoi(&p[0]) - 1900; - tm.tm_mon = atoi(&p[5]) - 1; - tm.tm_mday = atoi(&p[8]); - if ( (p[10] == 'T') && (p[13] == ':') ) { - tm.tm_hour = atoi(&p[11]); - tm.tm_min = atoi(&p[14]); - } - return mktime(&tm); + free(RSSAggr->Item); } - /* hmm... try RFC822 date stamp format */ + FreeAsyncIOContents(&RSSAggr->IO); + memset(RSSAggr, 0, sizeof(rss_aggregator)); + free(RSSAggr); +} + +eNextState RSSAggregator_Terminate(AsyncIO *IO) +{ + rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data; - t = parsedate(p); - if (t > 0) return(t); + EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n"); - /* yeesh. ok, just return the current date and time. */ - return(time(NULL)); + StopCurlWatchers(IO); + UnlinkRSSAggregator(RSSAggr); + return eAbort; } +eNextState RSSAggregator_TerminateDB(AsyncIO *IO) +{ + rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data; + EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n"); -void rss_xml_start(void *data, const char *supplied_el, const char **attr) { - struct rss_item *ri = (struct rss_item *) data; - char el[256]; - char *sep = NULL; - /* Axe the namespace, we don't care about it */ - safestrncpy(el, supplied_el, sizeof el); - while (sep = strchr(el, ':'), sep) { - strcpy(el, ++sep); - } + StopDBWatchers(&RSSAggr->IO); + UnlinkRSSAggregator(RSSAggr); + return eAbort; +} - if (!strcasecmp(el, "item")) { - ++ri->item_tag_nesting; - - /* Initialize the feed item data structure */ - if (ri->guid != NULL) free(ri->guid); - ri->guid = NULL; - if (ri->title != NULL) free(ri->title); - ri->title = NULL; - if (ri->link != NULL) free(ri->link); - ri->link = NULL; - if (ri->author_or_creator != NULL) free(ri->author_or_creator); - ri->author_or_creator = NULL; - if (ri->description != NULL) free(ri->description); - ri->description = NULL; - - /* Throw away any existing character data */ - if (ri->chardata_len > 0) { - free(ri->chardata); - ri->chardata = 0; - ri->chardata_len = 0; - } - } +eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO) +{ + const char *pUrl; + rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data; + pUrl = IO->ConnectMe->PlainUrl; + if (pUrl == NULL) + pUrl = ""; + EVRSSC_syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.\n", pUrl); + StopCurlWatchers(IO); + UnlinkRSSAggregator(RSSAggr); + return eAbort; } -void rss_xml_end(void *data, const char *supplied_el) { - struct rss_item *ri = (struct rss_item *) data; - char el[256]; - char *sep = NULL; - - /* Axe the namespace, we don't care about it */ - safestrncpy(el, supplied_el, sizeof el); - while (sep = strchr(el, ':'), sep) { - strcpy(el, ++sep); +void AppendLink(StrBuf *Message, + StrBuf *link, + StrBuf *LinkTitle, + const char *Title) +{ + if (StrLength(link) > 0) + { + StrBufAppendBufPlain(Message, HKEY(""), 0); + if (StrLength(LinkTitle) > 0) + StrBufAppendBuf(Message, LinkTitle, 0); + else if ((Title != NULL) && !IsEmptyStr(Title)) + StrBufAppendBufPlain(Message, Title, -1, 0); + else + StrBufAppendBuf(Message, link, 0); + StrBufAppendBufPlain(Message, HKEY("
\n"), 0); } +} - if ( (!strcasecmp(el, "title")) && (ri->item_tag_nesting == 0) && (ri->chardata != NULL) ) { - safestrncpy(ri->channel_title, ri->chardata, sizeof ri->channel_title); - striplt(ri->channel_title); - } - if ( (!strcasecmp(el, "guid")) && (ri->chardata != NULL) ) { - if (ri->guid != NULL) free(ri->guid); - striplt(ri->chardata); - ri->guid = strdup(ri->chardata); - } +void rss_format_item(networker_save_message *SaveMsg) +{ + StrBuf *Message; + int msglen = 0; - if ( (!strcasecmp(el, "title")) && (ri->chardata != NULL) ) { - if (ri->title != NULL) free(ri->title); - striplt(ri->chardata); - ri->title = strdup(ri->chardata); - } + if (SaveMsg->author_or_creator != NULL) { + + char *From; + StrBuf *Encoded = NULL; + int FromAt; + + From = html_to_ascii(ChrPtr(SaveMsg->author_or_creator), + StrLength(SaveMsg->author_or_creator), + 512, 0); + StrBufPlain(SaveMsg->author_or_creator, From, -1); + StrBufTrim(SaveMsg->author_or_creator); + free(From); + + FromAt = strchr(ChrPtr(SaveMsg->author_or_creator), '@') != NULL; + if (!FromAt && StrLength (SaveMsg->author_email) > 0) + { + StrBufRFC2047encode(&Encoded, SaveMsg->author_or_creator); + SaveMsg->Msg.cm_fields['A'] = SmashStrBuf(&Encoded); + SaveMsg->Msg.cm_fields['P'] = + SmashStrBuf(&SaveMsg->author_email); + } + else + { + if (FromAt) + { + SaveMsg->Msg.cm_fields['A'] = + SmashStrBuf(&SaveMsg->author_or_creator); + SaveMsg->Msg.cm_fields['P'] = + strdup(SaveMsg->Msg.cm_fields['A']); + } + else + { + StrBufRFC2047encode(&Encoded, + SaveMsg->author_or_creator); + SaveMsg->Msg.cm_fields['A'] = + SmashStrBuf(&Encoded); + SaveMsg->Msg.cm_fields['P'] = + strdup("rss@localhost"); - if ( (!strcasecmp(el, "link")) && (ri->chardata != NULL) ) { - if (ri->link != NULL) free(ri->link); - striplt(ri->chardata); - ri->link = strdup(ri->chardata); + } + } } - - if ( (!strcasecmp(el, "description")) && (ri->chardata != NULL) ) { - if (ri->description != NULL) free(ri->description); - ri->description = strdup(ri->chardata); + else { + SaveMsg->Msg.cm_fields['A'] = strdup("rss"); } - if ( ((!strcasecmp(el, "pubdate")) || (!strcasecmp(el, "date"))) && (ri->chardata != NULL) ) { - striplt(ri->chardata); - ri->pubdate = rdf_parsedate(ri->chardata); - } + SaveMsg->Msg.cm_fields['N'] = strdup(NODENAME); + if (SaveMsg->title != NULL) { + long len; + char *Sbj; + StrBuf *Encoded, *QPEncoded; + + QPEncoded = NULL; + StrBufSpaceToBlank(SaveMsg->title); + len = StrLength(SaveMsg->title); + Sbj = html_to_ascii(ChrPtr(SaveMsg->title), len, 512, 0); + len = strlen(Sbj); + if ((len > 0) && (Sbj[len - 1] == '\n')) + { + len --; + Sbj[len] = '\0'; + } + Encoded = NewStrBufPlain(Sbj, len); + free(Sbj); - if ( ((!strcasecmp(el, "author")) || (!strcasecmp(el, "creator"))) && (ri->chardata != NULL) ) { - if (ri->author_or_creator != NULL) free(ri->author_or_creator); - striplt(ri->chardata); - ri->author_or_creator = strdup(ri->chardata); - } + StrBufTrim(Encoded); + StrBufRFC2047encode(&QPEncoded, Encoded); - if (!strcasecmp(el, "item")) { - --ri->item_tag_nesting; - rss_save_item(ri); + SaveMsg->Msg.cm_fields['U'] = SmashStrBuf(&QPEncoded); + FreeStrBuf(&Encoded); } + if (SaveMsg->link == NULL) + SaveMsg->link = NewStrBufPlain(HKEY("")); - if ( (!strcasecmp(el, "rss")) || (!strcasecmp(el, "rdf")) ) { - CtdlLogPrintf(CTDL_DEBUG, "End of feed detected. Closing parser.\n"); - ri->done_parsing = 1; - } +#if 0 /* temporarily disable shorter urls. */ + SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] = + GetShorterUrls(SaveMsg->description); +#endif - if (ri->chardata_len > 0) { - free(ri->chardata); - ri->chardata = 0; - ri->chardata_len = 0; - } + msglen += 1024 + StrLength(SaveMsg->link) + StrLength(SaveMsg->description) ; -} + Message = NewStrBufPlain(NULL, msglen); + StrBufPlain(Message, HKEY( + "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n" + "\n")); +#if 0 /* disable shorter url for now. */ + SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message); +#endif + StrBufAppendBuf(Message, SaveMsg->description, 0); + StrBufAppendBufPlain(Message, HKEY("

\n"), 0); -/* - * This callback stores up the data which appears in between tags. - */ -void rss_xml_chardata(void *data, const XML_Char *s, int len) { - struct rss_item *ri = (struct rss_item *) data; - int old_len; - int new_len; - char *new_buffer; - - old_len = ri->chardata_len; - new_len = old_len + len; - new_buffer = realloc(ri->chardata, new_len + 1); - if (new_buffer != NULL) { - memcpy(&new_buffer[old_len], s, len); - new_buffer[new_len] = 0; - ri->chardata = new_buffer; - ri->chardata_len = new_len; - } -} + AppendLink(Message, SaveMsg->link, SaveMsg->linkTitle, NULL); + AppendLink(Message, SaveMsg->reLink, SaveMsg->reLinkTitle, "Reply to this"); + StrBufAppendBufPlain(Message, HKEY("\n"), 0); + SaveMsg->Message = Message; +} -/* - * Callback function for passing libcurl's output to expat for parsing - */ -size_t rss_libcurl_callback(void *ptr, size_t size, size_t nmemb, void *stream) +eNextState RSSSaveMessage(AsyncIO *IO) { - XML_Parse((XML_Parser)stream, ptr, (size * nmemb), 0); - return (size*nmemb); -} + long len; + const char *Key; + rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data; + rss_format_item(RSSAggr->ThisMsg); + RSSAggr->ThisMsg->Msg.cm_fields['M'] = + SmashStrBuf(&RSSAggr->ThisMsg->Message); -/* - * Begin a feed parse - */ -void rss_do_fetching(char *url, char *rooms) { - struct rss_item ri; - XML_Parser xp; + CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0); - CURL *curl; - CURLcode res; - char errmsg[1024] = ""; + /* write the uidl to the use table so we don't store this item again */ + cdb_store(CDB_USETABLE, + SKEY(RSSAggr->ThisMsg->MsgGUID), + &RSSAggr->ThisMsg->ut, + sizeof(struct UseTable) ); - CtdlLogPrintf(CTDL_DEBUG, "Fetching RSS feed <%s>\n", url); + if (GetNextHashPos(RSSAggr->Messages, + RSSAggr->Pos, + &len, &Key, + (void**) &RSSAggr->ThisMsg)) + return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry); + else + return eAbort; +} - curl = curl_easy_init(); - if (!curl) { - CtdlLogPrintf(CTDL_ALERT, "Unable to initialize libcurl.\n"); - return; - } +eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO) +{ + const char *Key; + long len; + struct cdbdata *cdbut; + rss_aggregator *Ctx = (rss_aggregator *) IO->Data; - xp = XML_ParserCreateNS("UTF-8", ':'); - if (!xp) { - CtdlLogPrintf(CTDL_ALERT, "Cannot create XML parser!\n"); - curl_easy_cleanup(curl); - return; - } + /* Find out if we've already seen this item */ + strcpy(Ctx->ThisMsg->ut.ut_msgid, + ChrPtr(Ctx->ThisMsg->MsgGUID)); /// TODO + Ctx->ThisMsg->ut.ut_timestamp = time(NULL); + + cdbut = cdb_fetch(CDB_USETABLE, SKEY(Ctx->ThisMsg->MsgGUID)); +#ifndef DEBUG_RSS + if (cdbut != NULL) { + /* Item has already been seen */ + EVRSSC_syslog(LOG_DEBUG, + "%s has already been seen\n", + ChrPtr(Ctx->ThisMsg->MsgGUID)); + cdb_free(cdbut); - curl_easy_setopt(curl, CURLOPT_URL, url); - curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 0); - curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 0); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, xp); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, rss_libcurl_callback); - curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, errmsg); - curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1); -#ifdef CURLOPT_HTTP_CONTENT_DECODING - curl_easy_setopt(curl, CURLOPT_HTTP_CONTENT_DECODING, 1); - curl_easy_setopt(curl, CURLOPT_ENCODING, ""); + /* rewrite the record anyway, to update the timestamp */ + cdb_store(CDB_USETABLE, + SKEY(Ctx->ThisMsg->MsgGUID), + &Ctx->ThisMsg->ut, sizeof(struct UseTable) ); + + if (GetNextHashPos(Ctx->Messages, + Ctx->Pos, + &len, &Key, + (void**) &Ctx->ThisMsg)) + return NextDBOperation( + IO, + RSS_FetchNetworkUsetableEntry); + else + return eAbort; + } + else #endif - curl_easy_setopt(curl, CURLOPT_USERAGENT, CITADEL); - curl_easy_setopt(curl, CURLOPT_TIMEOUT, 180); /* die after 180 seconds */ - if (!IsEmptyStr(config.c_ip_addr)) { - curl_easy_setopt(curl, CURLOPT_INTERFACE, config.c_ip_addr); + { + NextDBOperation(IO, RSSSaveMessage); + return eSendMore; } +} - memset(&ri, 0, sizeof(struct rss_item)); - ri.roomlist = rooms; - XML_SetElementHandler(xp, rss_xml_start, rss_xml_end); - XML_SetCharacterDataHandler(xp, rss_xml_chardata); - XML_SetUserData(xp, &ri); +eNextState RSSAggregator_AnalyseReply(AsyncIO *IO) +{ + struct UseTable ut; + u_char rawdigest[MD5_DIGEST_LEN]; + struct MD5Context md5context; + StrBuf *guid; + struct cdbdata *cdbut; + rss_aggregator *Ctx = (rss_aggregator *) IO->Data; - if (CtdlThreadCheckStop()) + if (IO->HttpReq.httpcode != 200) { - XML_ParserFree(xp); - curl_easy_cleanup(curl); - return; - } - - if (CtdlThreadCheckStop()) - goto shutdown ; + StrBuf *ErrMsg; + long lens[2]; + const char *strs[2]; - res = curl_easy_perform(curl); - if (res) { - CtdlLogPrintf(CTDL_ALERT, "libcurl error %d: %s\n", res, errmsg); + ErrMsg = NewStrBuf(); + EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n", + IO->HttpReq.httpcode); + + strs[0] = ChrPtr(Ctx->Url); + lens[0] = StrLength(Ctx->Url); + + strs[1] = ChrPtr(Ctx->rooms); + lens[1] = StrLength(Ctx->rooms); + StrBufPrintf(ErrMsg, + "Error while RSS-Aggregation Run of %s\n" + " need a 200, got a %ld !\n" + " Response text was: \n" + " \n %s\n", + ChrPtr(Ctx->Url), + IO->HttpReq.httpcode, + ChrPtr(IO->HttpReq.ReplyData)); + CtdlAideFPMessage( + ChrPtr(ErrMsg), + "RSS Aggregation run failure", + 2, strs, (long*) &lens); + FreeStrBuf(&ErrMsg); + return eAbort; } - if (CtdlThreadCheckStop()) - goto shutdown ; - - if (ri.done_parsing == 0) XML_Parse(xp, "", 0, 1); - - -shutdown: - curl_easy_cleanup(curl); - XML_ParserFree(xp); - - /* Free the feed item data structure */ - if (ri.guid != NULL) free(ri.guid); - ri.guid = NULL; - if (ri.title != NULL) free(ri.title); - ri.title = NULL; - if (ri.link != NULL) free(ri.link); - ri.link = NULL; - if (ri.author_or_creator != NULL) free(ri.author_or_creator); - ri.author_or_creator = NULL; - if (ri.description != NULL) free(ri.description); - ri.description = NULL; - if (ri.chardata_len > 0) { - free(ri.chardata); - ri.chardata = 0; - ri.chardata_len = 0; + MD5Init(&md5context); + + MD5Update(&md5context, + (const unsigned char*)SKEY(IO->HttpReq.ReplyData)); + + MD5Update(&md5context, + (const unsigned char*)SKEY(Ctx->Url)); + + MD5Final(rawdigest, &md5context); + guid = NewStrBufPlain(NULL, + MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/); + StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN); + StrBufAppendBufPlain(guid, HKEY("_rssFM"), 0); + if (StrLength(guid) > 40) + StrBufCutAt(guid, 40, NULL); + /* Find out if we've already seen this item */ + memcpy(ut.ut_msgid, SKEY(guid)); + ut.ut_timestamp = time(NULL); + + cdbut = cdb_fetch(CDB_USETABLE, SKEY(guid)); +#ifndef DEBUG_RSS + if (cdbut != NULL) { + /* Item has already been seen */ + EVRSSC_syslog(LOG_DEBUG, + "%s has already been seen\n", + ChrPtr(Ctx->Url)); + cdb_free(cdbut); } + + /* rewrite the record anyway, to update the timestamp */ + cdb_store(CDB_USETABLE, + SKEY(guid), + &ut, sizeof(struct UseTable) ); + FreeStrBuf(&guid); + if (cdbut != NULL) return eAbort; +#endif + return RSSAggregator_ParseReply(IO); } +eNextState RSSAggregator_FinishHttp(AsyncIO *IO) +{ + return QueueDBOperation(IO, RSSAggregator_AnalyseReply); +} + +/* + * Begin a feed parse + */ +int rss_do_fetching(rss_aggregator *RSSAggr) +{ + AsyncIO *IO = &RSSAggr->IO; + rss_item *ri; + time_t now; + + now = time(NULL); + + if ((RSSAggr->next_poll != 0) && (now < RSSAggr->next_poll)) + return 0; + + ri = (rss_item*) malloc(sizeof(rss_item)); + memset(ri, 0, sizeof(rss_item)); + RSSAggr->Item = ri; + + if (! InitcURLIOStruct(&RSSAggr->IO, + RSSAggr, + "Citadel RSS Client", + RSSAggregator_FinishHttp, + RSSAggregator_Terminate, + RSSAggregator_TerminateDB, + RSSAggregator_ShutdownAbort)) + { + EVRSSCM_syslog(LOG_ALERT, "Unable to initialize libcurl.\n"); + return 0; + } + + safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host, + ChrPtr(RSSAggr->Url), + sizeof(((CitContext*)RSSAggr->IO.CitContext)->cs_host)); + + EVRSSC_syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(RSSAggr->Url)); + ParseURL(&RSSAggr->IO.ConnectMe, RSSAggr->Url, 80); + CurlPrepareURL(RSSAggr->IO.ConnectMe); + + QueueCurlContext(&RSSAggr->IO); + return 1; +} /* * Scan a room's netconfig to determine whether it is requesting any RSS feeds */ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) { + StrBuf *CfgData=NULL; + StrBuf *CfgType; + StrBuf *Line; + rss_room_counter *Count = NULL; + struct stat statbuf; char filename[PATH_MAX]; - char buf[1024]; - char instr[32]; - FILE *fp; - char feedurl[256]; - struct rssnetcfg *rncptr = NULL; - struct rssnetcfg *use_this_rncptr = NULL; - int len = 0; - char *ptr = NULL; + int fd; + int Done; + rss_aggregator *RSSAggr = NULL; + rss_aggregator *use_this_RSSAggr = NULL; + void *vptr; + const char *CfgPtr, *lPtr; + const char *Err; + + pthread_mutex_lock(&RSSQueueMutex); + if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr)) + { + EVRSSQ_syslog(LOG_DEBUG, + "rssclient: [%ld] %s already in progress.\n", + qrbuf->QRnumber, + qrbuf->QRname); + pthread_mutex_unlock(&RSSQueueMutex); + return; + } + pthread_mutex_unlock(&RSSQueueMutex); assoc_file_name(filename, sizeof filename, qrbuf, ctdl_netcfg_dir); - if (CtdlThreadCheckStop()) + if (server_shutting_down) return; - + /* Only do net processing for rooms that have netconfigs */ - fp = fopen(filename, "r"); - if (fp == NULL) { + fd = open(filename, 0); + if (fd <= 0) { + /* syslog(LOG_DEBUG, + "rssclient: %s no config.\n", + qrbuf->QRname); */ return; } - while (fgets(buf, sizeof buf, fp) != NULL && !CtdlThreadCheckStop()) { - buf[strlen(buf)-1] = 0; - - extract_token(instr, buf, 0, '|', sizeof instr); - if (!strcasecmp(instr, "rssclient")) { + if (server_shutting_down) + return; - use_this_rncptr = NULL; + if (fstat(fd, &statbuf) == -1) { + EVRSSQ_syslog(LOG_DEBUG, + "ERROR: could not stat configfile '%s' - %s\n", + filename, + strerror(errno)); + return; + } - extract_token(feedurl, buf, 1, '|', sizeof feedurl); + if (server_shutting_down) + return; - /* If any other rooms have requested the same feed, then we will just add this - * room to the target list for that client request. - */ - for (rncptr=rnclist; rncptr!=NULL; rncptr=rncptr->next) { - if (!strcmp(rncptr->url, feedurl)) { - use_this_rncptr = rncptr; - } - } + CfgData = NewStrBufPlain(NULL, statbuf.st_size + 1); - /* Otherwise create a new client request */ - if (use_this_rncptr == NULL) { - rncptr = (struct rssnetcfg *) malloc(sizeof(struct rssnetcfg)); - if (rncptr != NULL) { - rncptr->next = rnclist; - safestrncpy(rncptr->url, feedurl, sizeof rncptr->url); - rncptr->rooms = NULL; - rnclist = rncptr; - use_this_rncptr = rncptr; - } - } + if (StrBufReadBLOB(CfgData, &fd, 1, statbuf.st_size, &Err) < 0) { + close(fd); + FreeStrBuf(&CfgData); + EVRSSQ_syslog(LOG_ERR, "ERROR: reading config '%s' - %s
\n", + filename, strerror(errno)); + return; + } + close(fd); + if (server_shutting_down) + return; - /* Add the room name to the request */ - if (use_this_rncptr != NULL) { - if (use_this_rncptr->rooms == NULL) { - rncptr->rooms = strdup(qrbuf->QRname); + CfgPtr = NULL; + CfgType = NewStrBuf(); + Line = NewStrBufPlain(NULL, StrLength(CfgData)); + Done = 0; + while (!Done) + { + Done = StrBufSipLine(Line, CfgData, &CfgPtr) == 0; + if (StrLength(Line) > 0) + { + lPtr = NULL; + StrBufExtract_NextToken(CfgType, Line, &lPtr, '|'); + if (!strcasecmp("rssclient", ChrPtr(CfgType))) + { + if (Count == NULL) + { + Count = malloc( + sizeof(rss_room_counter)); + Count->count = 0; } - else { - len = strlen(use_this_rncptr->rooms) + strlen(qrbuf->QRname) + 5; - ptr = realloc(use_this_rncptr->rooms, len); - if (ptr != NULL) { - strcat(ptr, "|"); - strcat(ptr, qrbuf->QRname); - use_this_rncptr->rooms = ptr; + Count->count ++; + RSSAggr = (rss_aggregator *) malloc( + sizeof(rss_aggregator)); + + memset (RSSAggr, 0, sizeof(rss_aggregator)); + RSSAggr->QRnumber = qrbuf->QRnumber; + RSSAggr->roomlist_parts = 1; + RSSAggr->Url = NewStrBuf(); + + StrBufExtract_NextToken(RSSAggr->Url, + Line, + &lPtr, + '|'); + + pthread_mutex_lock(&RSSQueueMutex); + GetHash(RSSFetchUrls, + SKEY(RSSAggr->Url), + &vptr); + + use_this_RSSAggr = (rss_aggregator *)vptr; + if (use_this_RSSAggr != NULL) + { + long *QRnumber; + StrBufAppendBufPlain( + use_this_RSSAggr->rooms, + qrbuf->QRname, + -1, 0); + if (use_this_RSSAggr->roomlist_parts==1) + { + use_this_RSSAggr->OtherQRnumbers + = NewHash(1, lFlathash); } + QRnumber = (long*)malloc(sizeof(long)); + *QRnumber = qrbuf->QRnumber; + Put(use_this_RSSAggr->OtherQRnumbers, + LKEY(qrbuf->QRnumber), + QRnumber, + NULL); + use_this_RSSAggr->roomlist_parts++; + + pthread_mutex_unlock(&RSSQueueMutex); + + FreeStrBuf(&RSSAggr->Url); + free(RSSAggr); + RSSAggr = NULL; + continue; } - } - } + pthread_mutex_unlock(&RSSQueueMutex); - } + RSSAggr->ItemType = RSS_UNSET; + + RSSAggr->rooms = NewStrBufPlain( + qrbuf->QRname, -1); - fclose(fp); + pthread_mutex_lock(&RSSQueueMutex); + Put(RSSFetchUrls, + SKEY(RSSAggr->Url), + RSSAggr, + DeleteRssCfg); + + pthread_mutex_unlock(&RSSQueueMutex); + } + } + } + if (Count != NULL) + { + Count->QRnumber = qrbuf->QRnumber; + pthread_mutex_lock(&RSSQueueMutex); + EVRSSQ_syslog(LOG_DEBUG, "client: [%ld] %s now starting.\n", + qrbuf->QRnumber, qrbuf->QRname); + Put(RSSQueueRooms, LKEY(qrbuf->QRnumber), Count, NULL); + pthread_mutex_unlock(&RSSQueueMutex); + } + FreeStrBuf(&CfgData); + FreeStrBuf(&CfgType); + FreeStrBuf(&Line); } /* * Scan for rooms that have RSS client requests configured */ -void *rssclient_scan(void *args) { - static time_t last_run = 0L; - static int doing_rssclient = 0; - struct rssnetcfg *rptr = NULL; - struct CitContext rssclientCC; - - /* Give this thread its own private CitContext */ - CtdlFillSystemContext(&rssclientCC, "rssclient"); - citthread_setspecific(MyConKey, (void *)&rssclientCC ); - - CtdlThreadAllocTSD(); +void rssclient_scan(void) { + int RSSRoomCount, RSSCount; + rss_aggregator *rptr = NULL; + void *vrptr = NULL; + HashPos *it; + long len; + const char *Key; + time_t now = time(NULL); + + /* Run no more than once every 15 minutes. */ + if ((now - last_run) < 900) { + EVRSSQ_syslog(LOG_DEBUG, + "Client: polling interval not yet reached; last run was %ldm%lds ago", + ((now - last_run) / 60), + ((now - last_run) % 60) + ); + return; + } /* - * This is a simple concurrency check to make sure only one rssclient run - * is done at a time. We could do this with a mutex, but since we - * don't really require extremely fine granularity here, we'll do it - * with a static variable instead. + * This is a simple concurrency check to make sure only one rssclient + * run is done at a time. */ - if (doing_rssclient) return NULL; - doing_rssclient = 1; - - CtdlLogPrintf(CTDL_DEBUG, "rssclient started\n"); - ForEachRoom(rssclient_scan_room, NULL); - - while (rnclist != NULL && !CtdlThreadCheckStop()) { - rss_do_fetching(rnclist->url, rnclist->rooms); - rptr = rnclist; - rnclist = rnclist->next; - if (rptr->rooms != NULL) free(rptr->rooms); - free(rptr); + pthread_mutex_lock(&RSSQueueMutex); + RSSCount = GetCount(RSSFetchUrls); + RSSRoomCount = GetCount(RSSQueueRooms); + pthread_mutex_unlock(&RSSQueueMutex); + + if ((RSSRoomCount > 0) || (RSSCount > 0)) { + EVRSSQ_syslog(LOG_DEBUG, + "rssclient: concurrency check failed; %d rooms and %d url's are queued", + RSSRoomCount, RSSCount + ); + return; } - CtdlLogPrintf(CTDL_DEBUG, "rssclient ended\n"); - last_run = time(NULL); - doing_rssclient = 0; - if (!CtdlThreadCheckStop()) - CtdlThreadSchedule ("RSS Client", CTDLTHREAD_BIGSTACK, rssclient_scan, NULL, last_run + config.c_net_freq); - else - CtdlLogPrintf(CTDL_DEBUG, "rssclient: Task STOPPED.\n"); - return NULL; + become_session(&rss_CC); + EVRSSQM_syslog(LOG_DEBUG, "rssclient started\n"); + CtdlForEachRoom(rssclient_scan_room, NULL); + + pthread_mutex_lock(&RSSQueueMutex); + + it = GetNewHashPos(RSSFetchUrls, 0); + while (!server_shutting_down && + GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && + (vrptr != NULL)) { + rptr = (rss_aggregator *)vrptr; + if (!rss_do_fetching(rptr)) + UnlinkRSSAggregator(rptr); + } + DeleteHashPos(&it); + pthread_mutex_unlock(&RSSQueueMutex); + + EVRSSQM_syslog(LOG_DEBUG, "rssclient ended\n"); + return; } +void rss_cleanup(void) +{ + /* citthread_mutex_destroy(&RSSQueueMutex); TODO */ + DeleteHash(&RSSFetchUrls); + DeleteHash(&RSSQueueRooms); +} + +void LogDebugEnableRSSClient(const int n) +{ + RSSClientDebugEnabled = n; +} CTDL_MODULE_INIT(rssclient) { if (threading) { - CtdlLogPrintf(CTDL_INFO, "%s\n", curl_version()); - CtdlThreadSchedule ("RSS Client", CTDLTHREAD_BIGSTACK, rssclient_scan, NULL, 0); + CtdlFillSystemContext(&rss_CC, "rssclient"); + pthread_mutex_init(&RSSQueueMutex, NULL); + RSSQueueRooms = NewHash(1, lFlathash); + RSSFetchUrls = NewHash(1, NULL); + syslog(LOG_INFO, "%s\n", curl_version()); + CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER, PRIO_AGGR + 300); + CtdlRegisterEVCleanupHook(rss_cleanup); + CtdlRegisterDebugFlagHook(HKEY("rssclient"), LogDebugEnableRSSClient, &RSSClientDebugEnabled); } - /* return our Subversion id for the Log */ - return "$Id$"; + return "rssclient"; }