X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Frssclient%2Fserv_rssclient.c;h=8a3ba63549a5aadacfa2bfd88d0cc25138541a01;hb=f74fed30f3d655d3f229628b6b998240fa6494be;hp=24e3d6097f1c1c29aa5deb2001546e721a9e0dde;hpb=f6877197c739711b051826923760edec66006376;p=citadel.git diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c index 24e3d6097..8a3ba6354 100644 --- a/citadel/modules/rssclient/serv_rssclient.c +++ b/citadel/modules/rssclient/serv_rssclient.c @@ -1,21 +1,21 @@ /* * Bring external RSS feeds into rooms. * - * Copyright (c) 2007-2010 by the citadel.org team + * Copyright (c) 2007-2012 by the citadel.org team * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 3 of the License, or - * (at your option) any later version. + * 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 + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.See the * GNU General Public License for more details. * - * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + * + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA02111-1307USA */ #include @@ -27,9 +27,9 @@ # include #else # if HAVE_SYS_TIME_H -# include +#include # else -# include +#include # endif #endif @@ -66,29 +66,13 @@ 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 to be locked too. */ - -eNextState RSSAggregatorTerminate(AsyncIO *IO); +HashList *RSSFetchUrls = NULL; /*->rss_aggregator;->RefCount access locked*/ +eNextState RSSAggregator_Terminate(AsyncIO *IO); +eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO); struct CitContext rss_CC; struct rssnetcfg *rnclist = NULL; -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); - } -} void DeleteRoomReference(long QRnumber) @@ -101,21 +85,22 @@ void DeleteRoomReference(long QRnumber) At = GetNewHashPos(RSSQueueRooms, 0); - GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At); - GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData); - if (vData != NULL) + if (GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At)) { - pRoomC = (rss_room_counter *) vData; - pRoomC->count --; - if (pRoomC->count == 0) - DeleteEntryFromHash(RSSQueueRooms, At); + GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData); + if (vData != NULL) + { + pRoomC = (rss_room_counter *) vData; + pRoomC->count --; + if (pRoomC->count == 0) + DeleteEntryFromHash(RSSQueueRooms, At); + } } DeleteHashPos(&At); } void UnlinkRooms(rss_aggregator *Cfg) { - DeleteRoomReference(Cfg->QRnumber); if (Cfg->OtherQRnumbers != NULL) { @@ -125,15 +110,16 @@ void UnlinkRooms(rss_aggregator *Cfg) void *vData; At = GetNewHashPos(Cfg->OtherQRnumbers, 0); - while (GetNextHashPos(Cfg->OtherQRnumbers, At, &HKLen, &HK, &vData) && + while (! server_shutting_down && + GetNextHashPos(Cfg->OtherQRnumbers, + At, + &HKLen, &HK, + &vData) && (vData != NULL)) { long *lData = (long*) vData; DeleteRoomReference(*lData); } -/* - if (server_shutting_down) - break; / * TODO */ DeleteHashPos(&At); } @@ -146,49 +132,71 @@ void UnlinkRSSAggregator(rss_aggregator *Cfg) UnlinkRooms(Cfg); At = GetNewHashPos(RSSFetchUrls, 0); - if (GetHashPosFromKey(RSSFetchUrls, SKEY(Cfg->Url), At) == 0) + if (GetHashPosFromKey(RSSFetchUrls, SKEY(Cfg->Url), At)) { DeleteEntryFromHash(RSSFetchUrls, At); } DeleteHashPos(&At); last_run = time(NULL); } -/* -eNextState FreeNetworkSaveMessage (AsyncIO *IO) + +void DeleteRssCfg(void *vptr) { - networker_save_message *Ctx = (networker_save_message *) IO->Data; + rss_aggregator *RSSAggr = (rss_aggregator *)vptr; + AsyncIO *IO = &RSSAggr->IO; + EVM_syslog(LOG_DEBUG, "RSS: destroying\n"); - pthread_mutex_lock(&RSSQueueMutex); - Ctx->Cfg->RefCount --; + FreeStrBuf(&RSSAggr->Url); + FreeStrBuf(&RSSAggr->rooms); + FreeStrBuf(&RSSAggr->CData); + FreeStrBuf(&RSSAggr->Key); + DeleteHash(&RSSAggr->OtherQRnumbers); + + DeleteHashPos (&RSSAggr->Pos); + DeleteHash (&RSSAggr->Messages); + if (RSSAggr->recp.recp_room != NULL) + free(RSSAggr->recp.recp_room); - if (Ctx->Cfg->RefCount == 0) + + if (RSSAggr->Item != NULL) { - UnlinkRSSAggregator(Ctx->Cfg); + flush_rss_item(RSSAggr->Item); + free(RSSAggr->Item); } - pthread_mutex_unlock(&RSSQueueMutex); - CtdlFreeMessage(Ctx->Msg); - free_recipients(Ctx->recp); - FreeStrBuf(&Ctx->Message); - FreeStrBuf(&Ctx->MsgGUID); - ((struct CitContext*)IO->CitContext)->state = CON_IDLE; - ((struct CitContext*)IO->CitContext)->kill_me = 1; - free(Ctx); - last_run = time(NULL); + FreeAsyncIOContents(&RSSAggr->IO); + free(RSSAggr); +} + +eNextState RSSAggregator_Terminate(AsyncIO *IO) +{ + rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data; + + EVM_syslog(LOG_DEBUG, "RSS: Terminating.\n"); + + + UnlinkRSSAggregator(RSSAggr); return eAbort; } -*/ -void FreeNetworkSaveMessage (void *vMsg) + +eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO) { - networker_save_message *Msg = (networker_save_message *) vMsg; + const char *pUrl; + rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data; + + pUrl = IO->ConnectMe->PlainUrl; + if (pUrl == NULL) + pUrl = ""; - CtdlFreeMessage(Msg->Msg); - FreeStrBuf(&Msg->Message); - FreeStrBuf(&Msg->MsgGUID); - free(Msg); + EV_syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.\n", pUrl); + + + UnlinkRSSAggregator(RSSAggr); + return eAbort; } + eNextState AbortNetworkSaveMessage (AsyncIO *IO) { return eAbort; ///TODO @@ -198,16 +206,23 @@ eNextState RSSSaveMessage(AsyncIO *IO) { long len; const char *Key; - rss_aggregator *Ctx = (rss_aggregator *) IO->Data; + rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data; - Ctx->ThisMsg->Msg->cm_fields['M'] = SmashStrBuf(&Ctx->ThisMsg->Message); + RSSAggr->ThisMsg->Msg.cm_fields['M'] = + SmashStrBuf(&RSSAggr->ThisMsg->Message); - CtdlSubmitMsg(Ctx->ThisMsg->Msg, &Ctx->recp, NULL, 0); + CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0); /* write the uidl to the use table so we don't store this item again */ - 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)) + cdb_store(CDB_USETABLE, + SKEY(RSSAggr->ThisMsg->MsgGUID), + &RSSAggr->ThisMsg->ut, + sizeof(struct UseTable) ); + + if (GetNextHashPos(RSSAggr->Messages, + RSSAggr->Pos, + &len, &Key, + (void**) &RSSAggr->ThisMsg)) return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry); else return eAbort; @@ -220,25 +235,32 @@ eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO) struct cdbdata *cdbut; rss_aggregator *Ctx = (rss_aggregator *) IO->Data; - /* Find out if we've already seen this item */ - strcpy(Ctx->ThisMsg->ut.ut_msgid, ChrPtr(Ctx->ThisMsg->MsgGUID)); /// TODO + 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 */ - EV_syslog(LOG_DEBUG, "%s has already been seen\n", ChrPtr(Ctx->ThisMsg->MsgGUID)); + EV_syslog(LOG_DEBUG, + "%s has already been seen\n", + ChrPtr(Ctx->ThisMsg->MsgGUID)); cdb_free(cdbut); /* rewrite the record anyway, to update the timestamp */ - cdb_store(CDB_USETABLE, - SKEY(Ctx->ThisMsg->MsgGUID), + 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); + if (GetNextHashPos(Ctx->Messages, + Ctx->Pos, + &len, &Key, + (void**) &Ctx->ThisMsg)) + return NextDBOperation( + IO, + RSS_FetchNetworkUsetableEntry); else return eAbort; } @@ -249,199 +271,6 @@ eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO) return eSendMore; } } -/* -void RSSAddSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf *MsgGUID, StrBuf *MessageBody, rss_aggregat *Cfg) -{ - networker_save_message *Ctx; - - pthread_mutex_lock(&RSSQueueMutex); - Cfg->RefCount ++; - pthread_mutex_unlock(&RSSQueueMutex); - - - Ctx = (networker_save_message *) malloc(sizeof(networker_save_message)); - memset(Ctx, 0, sizeof(networker_save_message)); - - Ctx->MsgGUID = MsgGUID; - Ctx->Message = MessageBody; - Ctx->Msg = Msg; - Ctx->Cfg = Cfg; - Ctx->recp = recp; - Ctx->IO.Data = Ctx; - Ctx->IO.CitContext = CloneContext(&rss_CC); - Ctx->IO.Terminate = FreeNetworkSaveMessage; - Ctx->IO.ShutdownAbort = AbortNetworkSaveMessage; - QueueDBOperation(&Ctx->IO, RSS_FetchNetworkUsetableEntry); -} -*/ - -/* - * Commit a fetched and parsed RSS item to disk - */ -void rss_save_item(rss_item *ri, rss_aggregator *Cfg) -{ - - struct MD5Context md5context; - u_char rawdigest[MD5_DIGEST_LEN]; - struct CtdlMessage *msg; - int msglen = 0; - StrBuf *Message; - StrBuf *guid; - AsyncIO *IO = &Cfg->IO; - - int n; - - /* 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) { - StrBufSpaceToBlank(ri->guid); - StrBufTrim(ri->guid); - guid = NewStrBufPlain(HKEY("rss/")); - StrBufAppendBuf(guid, ri->guid, 0); - } - else { - MD5Init(&md5context); - if (ri->title != NULL) { - MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->title), StrLength(ri->title)); - } - if (ri->link != NULL) { - MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->link), StrLength(ri->link)); - } - MD5Final(rawdigest, &md5context); - guid = NewStrBufPlain(NULL, MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/); - StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN); - StrBufAppendBufPlain(guid, HKEY("_rss2ctdl"), 0); - } - - /* translate Item into message. */ - EVM_syslog(LOG_DEBUG, "RSS: translating item...\n"); - if (ri->description == NULL) ri->description = NewStrBufPlain(HKEY("")); - StrBufSpaceToBlank(ri->description); - 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; - - if (ri->guid != NULL) { - msg->cm_fields['E'] = strdup(ChrPtr(ri->guid)); - } - - if (ri->author_or_creator != NULL) { - char *From; - StrBuf *Encoded = NULL; - int FromAt; - - From = html_to_ascii(ChrPtr(ri->author_or_creator), - StrLength(ri->author_or_creator), - 512, 0); - StrBufPlain(ri->author_or_creator, From, -1); - StrBufTrim(ri->author_or_creator); - free(From); - - FromAt = strchr(ChrPtr(ri->author_or_creator), '@') != NULL; - if (!FromAt && StrLength (ri->author_email) > 0) - { - StrBufRFC2047encode(&Encoded, ri->author_or_creator); - msg->cm_fields['A'] = SmashStrBuf(&Encoded); - msg->cm_fields['P'] = SmashStrBuf(&ri->author_email); - } - else - { - if (FromAt) - { - msg->cm_fields['A'] = SmashStrBuf(&ri->author_or_creator); - msg->cm_fields['P'] = strdup(msg->cm_fields['A']); - } - else - { - StrBufRFC2047encode(&Encoded, ri->author_or_creator); - msg->cm_fields['A'] = SmashStrBuf(&Encoded); - msg->cm_fields['P'] = strdup("rss@localhost"); - - } - if (ri->pubdate <= 0) { - ri->pubdate = time(NULL); - } - } - } - else { - msg->cm_fields['A'] = strdup("rss"); - } - - msg->cm_fields['N'] = strdup(NODENAME); - if (ri->title != NULL) { - long len; - char *Sbj; - StrBuf *Encoded, *QPEncoded; - - QPEncoded = NULL; - StrBufSpaceToBlank(ri->title); - len = StrLength(ri->title); - Sbj = html_to_ascii(ChrPtr(ri->title), len, 512, 0); - len = strlen(Sbj); - if (Sbj[len - 1] == '\n') - { - len --; - Sbj[len] = '\0'; - } - Encoded = NewStrBufPlain(Sbj, len); - free(Sbj); - - StrBufTrim(Encoded); - StrBufRFC2047encode(&QPEncoded, Encoded); - - msg->cm_fields['U'] = SmashStrBuf(&QPEncoded); - FreeStrBuf(&Encoded); - } - msg->cm_fields['T'] = malloc(64); - snprintf(msg->cm_fields['T'], 64, "%ld", ri->pubdate); - if (ri->channel_title != NULL) { - if (StrLength(ri->channel_title) > 0) { - msg->cm_fields['O'] = strdup(ChrPtr(ri->channel_title)); - } - } - if (ri->link == NULL) - ri->link = NewStrBufPlain(HKEY("")); - -#if 0 /* temporarily disable shorter urls. */ - msg->cm_fields[TMP_SHORTER_URLS] = GetShorterUrls(ri->description); -#endif - - msglen += 1024 + StrLength(ri->link) + StrLength(ri->description) ; - - Message = NewStrBufPlain(NULL, StrLength(ri->description)); - - StrBufPlain(Message, HKEY( - "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n" - "\n")); -#if 0 /* disable shorter url for now. */ - msg->cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message); -#endif - StrBufAppendBuf(Message, ri->description, 0); - StrBufAppendBufPlain(Message, HKEY("

\n"), 0); - - AppendLink(Message, ri->link, ri->linkTitle, NULL); - AppendLink(Message, ri->reLink, ri->reLinkTitle, "Reply to this"); - StrBufAppendBufPlain(Message, HKEY("\n"), 0); - - - - networker_save_message *SaveMsg; - - SaveMsg = (networker_save_message *) malloc(sizeof(networker_save_message)); - memset(SaveMsg, 0, sizeof(networker_save_message)); - - SaveMsg->MsgGUID = guid; - SaveMsg->Message = Message; - SaveMsg->Msg = msg; - - n = GetCount(Cfg->Messages) + 1; - Put(Cfg->Messages, IKEY(n), SaveMsg, FreeNetworkSaveMessage); -} - - /* * Begin a feed parse @@ -449,11 +278,9 @@ void rss_save_item(rss_item *ri, rss_aggregator *Cfg) int rss_do_fetching(rss_aggregator *Cfg) { rss_item *ri; - time_t now; - AsyncIO *IO; - now = time(NULL); + now = time(NULL); if ((Cfg->next_poll != 0) && (now < Cfg->next_poll)) return 0; @@ -461,79 +288,28 @@ int rss_do_fetching(rss_aggregator *Cfg) ri = (rss_item*) malloc(sizeof(rss_item)); memset(ri, 0, sizeof(rss_item)); Cfg->Item = ri; - IO = &Cfg->IO; - IO->CitContext = CloneContext(&rss_CC); - IO->Data = Cfg; - - syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(Cfg->Url)); - ParseURL(&IO->ConnectMe, Cfg->Url, 80); - CurlPrepareURL(IO->ConnectMe); - - if (! evcurl_init(IO, -// Ctx, - NULL, - "Citadel RSS Client", - ParseRSSReply, - RSSAggregatorTerminate)) + if (! InitcURLIOStruct(&Cfg->IO, + Cfg, + "Citadel RSS Client", + RSSAggregator_ParseReply, + RSSAggregator_Terminate, + RSSAggregator_ShutdownAbort)) { - syslog(LOG_DEBUG, "Unable to initialize libcurl.\n"); + syslog(LOG_ALERT, "Unable to initialize libcurl.\n"); return 0; } - QueueCurlContext(IO); - return 1; -} - - -void DeleteRssCfg(void *vptr) -{ - rss_aggregator *rncptr = (rss_aggregator *)vptr; - AsyncIO *IO = &rncptr->IO; - EVM_syslog(LOG_DEBUG, "RSS: destroying\n"); - - FreeStrBuf(&rncptr->Url); - FreeStrBuf(&rncptr->rooms); - FreeStrBuf(&rncptr->CData); - FreeStrBuf(&rncptr->Key); - FreeStrBuf(&rncptr->IO.HttpReq.ReplyData); - DeleteHash(&rncptr->OtherQRnumbers); - FreeURL(&rncptr->IO.ConnectMe); - - DeleteHashPos (&rncptr->Pos); - DeleteHash (&rncptr->Messages); - if (rncptr->recp.recp_room != NULL) - free(rncptr->recp.recp_room); - - - if (rncptr->Item != NULL) - { - FreeStrBuf(&rncptr->Item->guid); - FreeStrBuf(&rncptr->Item->title); - FreeStrBuf(&rncptr->Item->link); - FreeStrBuf(&rncptr->Item->linkTitle); - FreeStrBuf(&rncptr->Item->reLink); - FreeStrBuf(&rncptr->Item->reLinkTitle); - FreeStrBuf(&rncptr->Item->description); - FreeStrBuf(&rncptr->Item->channel_title); - FreeStrBuf(&rncptr->Item->author_or_creator); - FreeStrBuf(&rncptr->Item->author_url); - FreeStrBuf(&rncptr->Item->author_email); - - free(rncptr->Item); - } - free(rncptr); -} - -eNextState RSSAggregatorTerminate(AsyncIO *IO) -{ - rss_aggregator *rncptr = (rss_aggregator *)IO->Data; - - EVM_syslog(LOG_DEBUG, "RSS: Terminating.\n"); + safestrncpy(((CitContext*)Cfg->IO.CitContext)->cs_host, + ChrPtr(Cfg->Url), + sizeof(((CitContext*)Cfg->IO.CitContext)->cs_host)); + syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(Cfg->Url)); + ParseURL(&Cfg->IO.ConnectMe, Cfg->Url, 80); + CurlPrepareURL(Cfg->IO.ConnectMe); - UnlinkRSSAggregator(rncptr); - return eAbort; + QueueCurlContext(&Cfg->IO); + return 1; } /* @@ -547,10 +323,10 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) rss_room_counter *Count = NULL; struct stat statbuf; char filename[PATH_MAX]; - int fd; + int fd; int Done; - rss_aggregator *rncptr = NULL; - rss_aggregator *use_this_rncptr = NULL; + rss_aggregator *RSSAggr = NULL; + rss_aggregator *use_this_RSSAggr = NULL; void *vptr; const char *CfgPtr, *lPtr; const char *Err; @@ -558,10 +334,10 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) pthread_mutex_lock(&RSSQueueMutex); if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr)) { - syslog(LOG_DEBUG, - "rssclient: [%ld] %s already in progress.\n", - qrbuf->QRnumber, - qrbuf->QRname); + syslog(LOG_DEBUG, + "rssclient: [%ld] %s already in progress.\n", + qrbuf->QRnumber, + qrbuf->QRname); pthread_mutex_unlock(&RSSQueueMutex); return; } @@ -571,11 +347,13 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) if (server_shutting_down) return; - + /* Only do net processing for rooms that have netconfigs */ fd = open(filename, 0); if (fd <= 0) { - //syslog(LOG_DEBUG, "rssclient: %s no config.\n", qrbuf->QRname); + /* syslog(LOG_DEBUG, + "rssclient: %s no config.\n", + qrbuf->QRname); */ return; } @@ -583,8 +361,10 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) return; if (fstat(fd, &statbuf) == -1) { - syslog(LOG_DEBUG, "ERROR: could not stat configfile '%s' - %s\n", - filename, strerror(errno)); + syslog(LOG_DEBUG, + "ERROR: could not stat configfile '%s' - %s\n", + filename, + strerror(errno)); return; } @@ -597,81 +377,103 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) close(fd); FreeStrBuf(&CfgData); syslog(LOG_DEBUG, "ERROR: reading config '%s' - %s
\n", - filename, strerror(errno)); + filename, strerror(errno)); return; } close(fd); if (server_shutting_down) return; - + 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))) + Done = StrBufSipLine(Line, CfgData, &CfgPtr) == 0; + if (StrLength(Line) > 0) { - if (Count == NULL) - { - Count = malloc(sizeof(rss_room_counter)); - Count->count = 0; - } - Count->count ++; - rncptr = (rss_aggregator *) malloc(sizeof(rss_aggregator)); - memset (rncptr, 0, sizeof(rss_aggregator)); - rncptr->roomlist_parts = 1; - rncptr->Url = NewStrBuf(); - StrBufExtract_NextToken(rncptr->Url, Line, &lPtr, '|'); - - pthread_mutex_lock(&RSSQueueMutex); - GetHash(RSSFetchUrls, SKEY(rncptr->Url), &vptr); - use_this_rncptr = (rss_aggregator *)vptr; - if (use_this_rncptr != NULL) - { - long *QRnumber; - StrBufAppendBufPlain(use_this_rncptr->rooms, - qrbuf->QRname, - -1, 0); - if (use_this_rncptr->roomlist_parts == 1) - { - use_this_rncptr->OtherQRnumbers = NewHash(1, lFlathash); - } - QRnumber = (long*)malloc(sizeof(long)); - *QRnumber = qrbuf->QRnumber; - Put(use_this_rncptr->OtherQRnumbers, LKEY(qrbuf->QRnumber), QRnumber, NULL); - use_this_rncptr->roomlist_parts++; - - pthread_mutex_unlock(&RSSQueueMutex); - - FreeStrBuf(&rncptr->Url); - free(rncptr); - rncptr = NULL; - continue; - } - pthread_mutex_unlock(&RSSQueueMutex); - - rncptr->ItemType = RSS_UNSET; - - rncptr->rooms = NewStrBufPlain(qrbuf->QRname, -1); - - pthread_mutex_lock(&RSSQueueMutex); - Put(RSSFetchUrls, SKEY(rncptr->Url), rncptr, DeleteRssCfg); - pthread_mutex_unlock(&RSSQueueMutex); + lPtr = NULL; + StrBufExtract_NextToken(CfgType, Line, &lPtr, '|'); + if (!strcasecmp("rssclient", ChrPtr(CfgType))) + { + if (Count == NULL) + { + Count = malloc( + sizeof(rss_room_counter)); + Count->count = 0; + } + 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); + + 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); - syslog(LOG_DEBUG, "rssclient: [%ld] %s now starting.\n", - qrbuf->QRnumber, qrbuf->QRname); + syslog(LOG_DEBUG, "rssclient: [%ld] %s now starting.\n", + qrbuf->QRnumber, qrbuf->QRname); Put(RSSQueueRooms, LKEY(qrbuf->QRnumber), Count, NULL); pthread_mutex_unlock(&RSSQueueMutex); } @@ -684,29 +486,29 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) * Scan for rooms that have RSS client requests configured */ void rssclient_scan(void) { - static int doing_rssclient = 0; rss_aggregator *rptr = NULL; void *vrptr = NULL; - HashPos *it; + HashPos *it; long len; const char *Key; + time_t now = time(NULL); /* Run no more than once every 15 minutes. */ - if ((time(NULL) - last_run) < 900) { + if ((now - last_run) < 900) { 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 + * 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. */ - if (doing_rssclient) return; - doing_rssclient = 1; + if ((GetCount(RSSQueueRooms) > 0) || (GetCount(RSSFetchUrls) > 0)) return; + become_session(&rss_CC); syslog(LOG_DEBUG, "rssclient started\n"); CtdlForEachRoom(rssclient_scan_room, NULL); @@ -714,7 +516,7 @@ void rssclient_scan(void) { it = GetNewHashPos(RSSFetchUrls, 0); while (!server_shutting_down && - GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && + GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && (vrptr != NULL)) { rptr = (rss_aggregator *)vrptr; if (!rss_do_fetching(rptr)) @@ -724,7 +526,6 @@ void rssclient_scan(void) { pthread_mutex_unlock(&RSSQueueMutex); syslog(LOG_DEBUG, "rssclient ended\n"); - doing_rssclient = 0; return; } @@ -746,7 +547,7 @@ CTDL_MODULE_INIT(rssclient) RSSFetchUrls = NewHash(1, NULL); syslog(LOG_INFO, "%s\n", curl_version()); CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER); - CtdlRegisterCleanupHook(rss_cleanup); + CtdlRegisterEVCleanupHook(rss_cleanup); } return "rssclient"; }