X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Frssclient%2Fserv_rssclient.c;h=24e3d6097f1c1c29aa5deb2001546e721a9e0dde;hb=f6877197c739711b051826923760edec66006376;hp=a01a92f1c46fefcbee702e9dd46768dd1065ecce;hpb=0149ea4b722c795dfdb9918d2281e859cf064ebe;p=citadel.git diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c index a01a92f1c..24e3d6097 100644 --- a/citadel/modules/rssclient/serv_rssclient.c +++ b/citadel/modules/rssclient/serv_rssclient.c @@ -62,6 +62,15 @@ #define TMP_SHORTER_URL_OFFSET 0xFE #define TMP_SHORTER_URLS 0xFD +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); + +struct CitContext rss_CC; struct rssnetcfg *rnclist = NULL; void AppendLink(StrBuf *Message, StrBuf *link, StrBuf *LinkTitle, const char *Title) @@ -81,49 +90,206 @@ void AppendLink(StrBuf *Message, StrBuf *link, StrBuf *LinkTitle, const char *Ti } } -void RSSSaveMessage(struct CtdlMessage *Msg, rss_item *ri, struct UseTable *ut) + +void DeleteRoomReference(long QRnumber) +{ + HashPos *At; + long HKLen; + const char *HK; + void *vData = NULL; + rss_room_counter *pRoomC; + + At = GetNewHashPos(RSSQueueRooms, 0); + + 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); + } + DeleteHashPos(&At); +} + +void UnlinkRooms(rss_aggregator *Cfg) +{ + + DeleteRoomReference(Cfg->QRnumber); + if (Cfg->OtherQRnumbers != NULL) + { + long HKLen; + const char *HK; + HashPos *At; + void *vData; + + At = GetNewHashPos(Cfg->OtherQRnumbers, 0); + while (GetNextHashPos(Cfg->OtherQRnumbers, At, &HKLen, &HK, &vData) && + (vData != NULL)) + { + long *lData = (long*) vData; + DeleteRoomReference(*lData); + } +/* + if (server_shutting_down) + break; / * TODO */ + + DeleteHashPos(&At); + } +} + +void UnlinkRSSAggregator(rss_aggregator *Cfg) +{ + HashPos *At; + + UnlinkRooms(Cfg); + + At = GetNewHashPos(RSSFetchUrls, 0); + if (GetHashPosFromKey(RSSFetchUrls, SKEY(Cfg->Url), At) == 0) + { + DeleteEntryFromHash(RSSFetchUrls, At); + } + DeleteHashPos(&At); + last_run = time(NULL); +} +/* +eNextState FreeNetworkSaveMessage (AsyncIO *IO) +{ + networker_save_message *Ctx = (networker_save_message *) IO->Data; + + pthread_mutex_lock(&RSSQueueMutex); + Ctx->Cfg->RefCount --; + + if (Ctx->Cfg->RefCount == 0) + { + UnlinkRSSAggregator(Ctx->Cfg); + + } + 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); + return eAbort; +} +*/ +void FreeNetworkSaveMessage (void *vMsg) +{ + networker_save_message *Msg = (networker_save_message *) vMsg; + + CtdlFreeMessage(Msg->Msg); + FreeStrBuf(&Msg->Message); + FreeStrBuf(&Msg->MsgGUID); + free(Msg); +} + +eNextState AbortNetworkSaveMessage (AsyncIO *IO) +{ + return eAbort; ///TODO +} + +eNextState RSSSaveMessage(AsyncIO *IO) { + long len; + const char *Key; + rss_aggregator *Ctx = (rss_aggregator *) IO->Data; - CtdlSubmitMsg(msg, recp, NULL, 0); - CtdlFreeMessage(msg); + Ctx->ThisMsg->Msg->cm_fields['M'] = SmashStrBuf(&Ctx->ThisMsg->Message); + + CtdlSubmitMsg(Ctx->ThisMsg->Msg, &Ctx->recp, NULL, 0); /* write the uidl to the use table so we don't store this item again */ - cdb_store(CDB_USETABLE, utmsgid, strlen(utmsgid), &ut, sizeof(struct UseTable) ); - free(ut); + 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; } +eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO) +{ + const char *Key; + long len; + 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 + 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)); + cdb_free(cdbut); + + /* rewrite the record anyway, to update the timestamp */ + cdb_store(CDB_USETABLE, + SKEY(Ctx->ThisMsg->MsgGUID), + &Ctx->ThisMsg->ut, sizeof(struct UseTable) ); -rss_save_msg(msg, recp) + if (GetNextHashPos(Ctx->Messages, Ctx->Pos, &len, &Key, (void**) &Ctx->ThisMsg)) + return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry); + else + return eAbort; + } + else +#endif + { + NextDBOperation(IO, RSSSaveMessage); + 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) +void rss_save_item(rss_item *ri, rss_aggregator *Cfg) { 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; StrBuf *Message; - AsyncIO *OtherIO; - - recp = (struct recptypes *) malloc(sizeof(struct recptypes)); - if (recp == NULL) return; - memset(recp, 0, sizeof(struct recptypes)); - memset(&ut, 0, sizeof(struct UseTable)); - recp->recp_room = strdup(ri->roomlist); - recp->num_room = num_tokens(ri->roomlist, '|'); - recp->recptypes_magic = RECPTYPES_MAGIC; + 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. @@ -131,7 +297,8 @@ void rss_save_item(rss_item *ri) if (ri->guid != NULL) { StrBufSpaceToBlank(ri->guid); StrBufTrim(ri->guid); - snprintf(utmsgid, sizeof utmsgid, "rss/%s", ChrPtr(ri->guid)); + guid = NewStrBufPlain(HKEY("rss/")); + StrBufAppendBuf(guid, ri->guid, 0); } else { MD5Init(&md5context); @@ -142,16 +309,13 @@ void rss_save_item(rss_item *ri) MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->link), StrLength(ri->link)); } MD5Final(rawdigest, &md5context); - for (i=0; idescription == NULL) ri->description = NewStrBufPlain(HKEY("")); StrBufSpaceToBlank(ri->description); msg = malloc(sizeof(struct CtdlMessage)); @@ -186,12 +350,19 @@ void rss_save_item(rss_item *ri) else { if (FromAt) - msg->cm_fields['P'] = SmashStrBuf(&ri->author_or_creator); + { + 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); } } } @@ -234,11 +405,9 @@ void rss_save_item(rss_item *ri) if (ri->link == NULL) ri->link = NewStrBufPlain(HKEY("")); - +#if 0 /* temporarily disable shorter urls. */ msg->cm_fields[TMP_SHORTER_URLS] = GetShorterUrls(ri->description); - - strcpy(ut->ut_msgid, utmsgid); - ut->ut_timestamp = time(NULL); +#endif msglen += 1024 + StrLength(ri->link) + StrLength(ri->description) ; @@ -247,7 +416,9 @@ void rss_save_item(rss_item *ri) 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); @@ -256,76 +427,46 @@ void rss_save_item(rss_item *ri) StrBufAppendBufPlain(Message, HKEY("\n"), 0); - msg->cm_fields[TMP_MSGDATA] = Message; - - - OtherIO = malloc(sizeof(AsyncIO)); - memset(OtherIO, 0, sizeof(AsyncIO)); - OtherIO->AsyncMsg = msg; - OtherIO->AsyncRcp = recp; - rss_save_msg(msg, recp); -// msg->cm_fields['M'] = SmashStrBuf(&Message); + networker_save_message *SaveMsg; - // TODO: reenable me ExpandShortUrls(ri->description); + 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; -/// free_recipients(recp); + n = GetCount(Cfg->Messages) + 1; + Put(Cfg->Messages, IKEY(n), SaveMsg, FreeNetworkSaveMessage); } - - /* Find out if we've already seen this item * / - - cdbut = cdb_fetch(CDB_USETABLE, utmsgid, strlen(utmsgid)); -#ifndef DEBUG_RSS - if (cdbut != NULL) { - /* Item has already been seen * / - CtdlLogPrintf(CTDL_DEBUG, "%s has already been seen\n", utmsgid); - cdb_free(cdbut); - - /* 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 -#endif - { -*/ - - /* * Begin a feed parse */ -void rss_do_fetching(rssnetcfg *Cfg) { - rsscollection *rssc; +int rss_do_fetching(rss_aggregator *Cfg) +{ rss_item *ri; time_t now; - - CURL *chnd; AsyncIO *IO; now = time(NULL); if ((Cfg->next_poll != 0) && (now < Cfg->next_poll)) - return; - + return 0; ri = (rss_item*) malloc(sizeof(rss_item)); - rssc = (rsscollection*) malloc(sizeof(rsscollection)); memset(ri, 0, sizeof(rss_item)); - memset(rssc, 0, sizeof(rsscollection)); - rssc->Item = ri; - rssc->Cfg = Cfg; - IO = &rssc->IO; - IO->CitContext = CloneContext(CC); - IO->Data = rssc; - ri->roomlist = Cfg->rooms; + Cfg->Item = ri; + IO = &Cfg->IO; + IO->CitContext = CloneContext(&rss_CC); + IO->Data = Cfg; - CtdlLogPrintf(CTDL_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(Cfg->Url)); + syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(Cfg->Url)); ParseURL(&IO->ConnectMe, Cfg->Url, 80); CurlPrepareURL(IO->ConnectMe); @@ -333,108 +474,227 @@ void rss_do_fetching(rssnetcfg *Cfg) { // Ctx, NULL, "Citadel RSS Client", - ParseRSSReply)) + ParseRSSReply, + RSSAggregatorTerminate)) { - CtdlLogPrintf(CTDL_ALERT, "Unable to initialize libcurl.\n"); -// goto abort; + syslog(LOG_DEBUG, "Unable to initialize libcurl.\n"); + return 0; } - chnd = IO->HttpReq.chnd; - evcurl_handle_start(IO); + 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"); + + + UnlinkRSSAggregator(rncptr); + return eAbort; +} + /* * 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]; - rssnetcfg *rncptr = NULL; - rssnetcfg *use_this_rncptr = NULL; - int len = 0; - char *ptr = NULL; + int fd; + int Done; + rss_aggregator *rncptr = NULL; + rss_aggregator *use_this_rncptr = NULL; + void *vptr; + const char *CfgPtr, *lPtr; + const char *Err; + + 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); + 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")) { - - use_this_rncptr = NULL; + if (server_shutting_down) + return; - extract_token(feedurl, buf, 1, '|', sizeof feedurl); + if (fstat(fd, &statbuf) == -1) { + syslog(LOG_DEBUG, "ERROR: could not stat configfile '%s' - %s\n", + filename, strerror(errno)); + 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(ChrPtr(rncptr->Url), feedurl)) { - use_this_rncptr = rncptr; - } - } + if (server_shutting_down) + return; - /* Otherwise create a new client request */ - if (use_this_rncptr == NULL) { - rncptr = (rssnetcfg *) malloc(sizeof(rssnetcfg)); - memset(rncptr, 0, sizeof(rssnetcfg)); - rncptr->ItemType = RSS_UNSET; - if (rncptr != NULL) { - rncptr->next = rnclist; - rncptr->Url = NewStrBufPlain(feedurl, -1); - rncptr->rooms = NULL; - rnclist = rncptr; - use_this_rncptr = rncptr; - } - } + CfgData = NewStrBufPlain(NULL, statbuf.st_size + 1); - /* Add the room name to the request */ - if (use_this_rncptr != NULL) { - if (use_this_rncptr->rooms == NULL) { - rncptr->rooms = strdup(qrbuf->QRname); - } - 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; - } - } - } + if (StrBufReadBLOB(CfgData, &fd, 1, statbuf.st_size, &Err) < 0) { + close(fd); + FreeStrBuf(&CfgData); + syslog(LOG_DEBUG, "ERROR: reading config '%s' - %s
\n", + 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))) + { + 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); } - + } } - - fclose(fp); - + if (Count != NULL) + { + Count->QRnumber = qrbuf->QRnumber; + pthread_mutex_lock(&RSSQueueMutex); + syslog(LOG_DEBUG, "rssclient: [%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) { - static time_t last_run = 0L; static int doing_rssclient = 0; - rssnetcfg *rptr = NULL; + rss_aggregator *rptr = NULL; + void *vrptr = NULL; + HashPos *it; + long len; + const char *Key; + + /* Run no more than once every 15 minutes. */ + if ((time(NULL) - last_run) < 900) { + return; + } /* * This is a simple concurrency check to make sure only one rssclient run @@ -444,31 +704,49 @@ void rssclient_scan(void) { */ if (doing_rssclient) return; doing_rssclient = 1; + if ((GetCount(RSSQueueRooms) > 0) || (GetCount(RSSFetchUrls) > 0)) + return; - CtdlLogPrintf(CTDL_DEBUG, "rssclient started\n"); + syslog(LOG_DEBUG, "rssclient started\n"); CtdlForEachRoom(rssclient_scan_room, NULL); - while (rnclist != NULL && !CtdlThreadCheckStop()) { - rss_do_fetching(rnclist); - rptr = rnclist; - rnclist = rnclist->next; - if (rptr->rooms != NULL) free(rptr->rooms); - free(rptr); + 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); - CtdlLogPrintf(CTDL_DEBUG, "rssclient ended\n"); - last_run = time(NULL); + syslog(LOG_DEBUG, "rssclient ended\n"); doing_rssclient = 0; return; } +void rss_cleanup(void) +{ + /* citthread_mutex_destroy(&RSSQueueMutex); TODO */ + DeleteHash(&RSSFetchUrls); + DeleteHash(&RSSQueueRooms); +} + CTDL_MODULE_INIT(rssclient) { if (threading) { - CtdlLogPrintf(CTDL_INFO, "%s\n", curl_version()); + 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); + CtdlRegisterCleanupHook(rss_cleanup); } return "rssclient"; }