X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Frssclient%2Fserv_rssclient.c;h=530bed03fe6f689b8bb96eb6af547744f980f33c;hb=c855d497545dad80942a194624c111a54cd1fdc7;hp=74f1992cbba458f3f4b1dacfbe9c0f8b3e086846;hpb=1dd7c0f770f23bbc4f65de406deb75eb059b417f;p=citadel.git diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c index 74f1992cb..530bed03f 100644 --- a/citadel/modules/rssclient/serv_rssclient.c +++ b/citadel/modules/rssclient/serv_rssclient.c @@ -62,9 +62,21 @@ #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 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) +void AppendLink(StrBuf *Message, + StrBuf *link, + StrBuf *LinkTitle, + const char *Title) { if (StrLength(link) > 0) { @@ -80,68 +92,148 @@ void AppendLink(StrBuf *Message, StrBuf *link, StrBuf *LinkTitle, const char *Ti StrBufAppendBufPlain(Message, HKEY("
\n"), 0); } } -typedef struct __networker_save_message { - AsyncIO IO; - struct CtdlMessage *Msg; - struct recptypes *recp; - StrBuf *MsgGUID; - StrBuf *Message; - struct UseTable ut; -} networker_save_message; -eNextState FreeNetworkSaveMessage (AsyncIO *IO) + +void DeleteRoomReference(long QRnumber) { - networker_save_message *Ctx = (networker_save_message *) IO->Data; + HashPos *At; + long HKLen; + const char *HK; + void *vData = NULL; + rss_room_counter *pRoomC; - CtdlFreeMessage(Ctx->Msg); - free_recipients(Ctx->recp); - FreeStrBuf(&Ctx->MsgGUID); - free(Ctx); - return eAbort; + At = GetNewHashPos(RSSQueueRooms, 0); + + 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); + } + } + 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 (! server_shutting_down && + GetNextHashPos(Cfg->OtherQRnumbers, + At, + &HKLen, &HK, + &vData) && + (vData != NULL)) + { + long *lData = (long*) vData; + DeleteRoomReference(*lData); + } + + DeleteHashPos(&At); + } +} + +void UnlinkRSSAggregator(rss_aggregator *Cfg) +{ + HashPos *At; + + UnlinkRooms(Cfg); + + At = GetNewHashPos(RSSFetchUrls, 0); + if (GetHashPosFromKey(RSSFetchUrls, SKEY(Cfg->Url), At)) + { + DeleteEntryFromHash(RSSFetchUrls, At); + } + DeleteHashPos(&At); + last_run = time(NULL); +} + +void FreeNetworkSaveMessage (void *vMsg) +{ + networker_save_message *Msg = (networker_save_message *) vMsg; + + CtdlFreeMessageContents(&Msg->Msg); + FreeStrBuf(&Msg->Message); + FreeStrBuf(&Msg->MsgGUID); + free(Msg); } eNextState AbortNetworkSaveMessage (AsyncIO *IO) { - return eAbort; ///TODO + return eAbort; ///TODO } eNextState RSSSaveMessage(AsyncIO *IO) { - networker_save_message *Ctx = (networker_save_message *) IO->Data; + long len; + const char *Key; + rss_aggregator *Ctx = (rss_aggregator *) IO->Data; - Ctx->Msg->cm_fields['M'] = SmashStrBuf(&Ctx->Message); + Ctx->ThisMsg->Msg.cm_fields['M'] = SmashStrBuf(&Ctx->ThisMsg->Message); - CtdlSubmitMsg(Ctx->Msg, Ctx->recp, NULL, 0); + 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, SKEY(Ctx->MsgGUID), &Ctx->ut, sizeof(struct UseTable) ); - - return eTerminateConnection; + 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; } -// TODO: relink me: ExpandShortUrls(ri->description); - -eNextState FetchNetworkUsetableEntry(AsyncIO *IO) +eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO) { + const char *Key; + long len; struct cdbdata *cdbut; - networker_save_message *Ctx = (networker_save_message *) IO->Data; + rss_aggregator *Ctx = (rss_aggregator *) IO->Data; /* Find out if we've already seen this item */ - strcpy(Ctx->ut.ut_msgid, ChrPtr(Ctx->MsgGUID)); /// TODO - Ctx->ut.ut_timestamp = time(NULL); + 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->MsgGUID)); + cdbut = cdb_fetch(CDB_USETABLE, SKEY(Ctx->ThisMsg->MsgGUID)); #ifndef DEBUG_RSS if (cdbut != NULL) { /* Item has already been seen */ - CtdlLogPrintf(CTDL_DEBUG, "%s has already been seen\n", ChrPtr(Ctx->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->MsgGUID), - &Ctx->ut, sizeof(struct UseTable) ); - return eTerminateConnection; + 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 @@ -150,46 +242,26 @@ eNextState FetchNetworkUsetableEntry(AsyncIO *IO) return eSendMore; } } -void RSSQueueSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf *MsgGUID, StrBuf *MessageBody) -{ - networker_save_message *Ctx; - - 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->recp = recp; - Ctx->IO.Data = Ctx; - Ctx->IO.CitContext = CloneContext(CC); - Ctx->IO.Terminate = FreeNetworkSaveMessage; - Ctx->IO.ShutdownAbort = AbortNetworkSaveMessage; - QueueDBOperation(&Ctx->IO, 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) { - + networker_save_message *SaveMsg; struct MD5Context md5context; u_char rawdigest[MD5_DIGEST_LEN]; - struct CtdlMessage *msg; - struct recptypes *recp = NULL; int msglen = 0; StrBuf *Message; StrBuf *guid; + AsyncIO *IO = &Cfg->IO; + int n; + + + SaveMsg = (networker_save_message *) malloc( + sizeof(networker_save_message)); + memset(SaveMsg, 0, sizeof(networker_save_message)); - 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. */ @@ -202,38 +274,39 @@ void rss_save_item(rss_item *ri) else { MD5Init(&md5context); if (ri->title != NULL) { - MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->title), StrLength(ri->title)); + MD5Update(&md5context, + (const unsigned char*)SKEY(ri->title)); } if (ri->link != NULL) { - MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->link), StrLength(ri->link)); + MD5Update(&md5context, + (const unsigned char*)SKEY(ri->link)); } MD5Final(rawdigest, &md5context); - guid = NewStrBufPlain(NULL, MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/); + 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. */ - CtdlLogPrintf(CTDL_DEBUG, "RSS: translating item...\n"); + 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; + SaveMsg->Msg.cm_magic = CTDLMESSAGE_MAGIC; + SaveMsg->Msg.cm_anon_type = MES_NORMAL; + SaveMsg->Msg.cm_format_type = FMT_RFC822; if (ri->guid != NULL) { - msg->cm_fields['E'] = strdup(ChrPtr(ri->guid)); + SaveMsg->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), + StrLength(ri->author_or_creator), 512, 0); StrBufPlain(ri->author_or_creator, From, -1); StrBufTrim(ri->author_or_creator); @@ -243,26 +316,39 @@ void rss_save_item(rss_item *ri) 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); + SaveMsg->Msg.cm_fields['A'] = SmashStrBuf(&Encoded); + SaveMsg->Msg.cm_fields['P'] = + SmashStrBuf(&ri->author_email); } else { if (FromAt) - msg->cm_fields['P'] = SmashStrBuf(&ri->author_or_creator); - else { - StrBufRFC2047encode(&Encoded, ri->author_or_creator); - msg->cm_fields['A'] = SmashStrBuf(&Encoded); - msg->cm_fields['P'] = strdup("rss@localhost"); + SaveMsg->Msg.cm_fields['A'] = + SmashStrBuf(&ri->author_or_creator); + SaveMsg->Msg.cm_fields['P'] = + strdup(SaveMsg->Msg.cm_fields['A']); + } + else + { + StrBufRFC2047encode(&Encoded, + ri->author_or_creator); + SaveMsg->Msg.cm_fields['A'] = + SmashStrBuf(&Encoded); + SaveMsg->Msg.cm_fields['P'] = + strdup("rss@localhost"); + + } + if (ri->pubdate <= 0) { + ri->pubdate = time(NULL); } } } else { - msg->cm_fields['A'] = strdup("rss"); + SaveMsg->Msg.cm_fields['A'] = strdup("rss"); } - msg->cm_fields['N'] = strdup(NODENAME); + SaveMsg->Msg.cm_fields['N'] = strdup(NODENAME); if (ri->title != NULL) { long len; char *Sbj; @@ -284,21 +370,23 @@ void rss_save_item(rss_item *ri) StrBufTrim(Encoded); StrBufRFC2047encode(&QPEncoded, Encoded); - msg->cm_fields['U'] = SmashStrBuf(&QPEncoded); + SaveMsg->Msg.cm_fields['U'] = SmashStrBuf(&QPEncoded); FreeStrBuf(&Encoded); } - msg->cm_fields['T'] = malloc(64); - snprintf(msg->cm_fields['T'], 64, "%ld", ri->pubdate); + SaveMsg->Msg.cm_fields['T'] = malloc(64); + snprintf(SaveMsg->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)); + SaveMsg->Msg.cm_fields['O'] = + strdup(ChrPtr(ri->channel_title)); } } - if (ri->link == NULL) + if (ri->link == NULL) ri->link = NewStrBufPlain(HKEY("")); #if 0 /* temporarily disable shorter urls. */ - msg->cm_fields[TMP_SHORTER_URLS] = GetShorterUrls(ri->description); + SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] = + GetShorterUrls(ri->description); #endif msglen += 1024 + StrLength(ri->link) + StrLength(ri->description) ; @@ -309,7 +397,7 @@ void rss_save_item(rss_item *ri) "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); + SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message); #endif StrBufAppendBuf(Message, ri->description, 0); StrBufAppendBufPlain(Message, HKEY("

\n"), 0); @@ -318,7 +406,11 @@ void rss_save_item(rss_item *ri) AppendLink(Message, ri->reLink, ri->reLinkTitle, "Reply to this"); StrBufAppendBufPlain(Message, HKEY("\n"), 0); - RSSQueueSaveMessage(msg, recp, guid, Message); + SaveMsg->MsgGUID = guid; + SaveMsg->Message = Message; + + n = GetCount(Cfg->Messages) + 1; + Put(Cfg->Messages, IKEY(n), SaveMsg, FreeNetworkSaveMessage); } @@ -326,187 +418,181 @@ void rss_save_item(rss_item *ri) /* * 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; - AsyncIO *IO; - now = time(NULL); + now = time(NULL); if ((Cfg->next_poll != 0) && (now < Cfg->next_poll)) - return; - Cfg->Attached = 1; + 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; - - - CtdlLogPrintf(CTDL_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)) + Cfg->Item = ri; + + if (! InitcURLIOStruct(&Cfg->IO, + Cfg, + "Citadel RSS Client", + RSSAggregator_ParseReply, + RSSAggregator_Terminate, + RSSAggregator_ShutdownAbort)) { - CtdlLogPrintf(CTDL_ALERT, "Unable to initialize libcurl.\n"); -// goto abort; + syslog(LOG_ALERT, "Unable to initialize libcurl.\n"); + return 0; } - evcurl_handle_start(IO); + 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); + + QueueCurlContext(&Cfg->IO); + return 1; } -citthread_mutex_t RSSQueueMutex; /* locks the access to the following vars: */ -HashList *RSSQueueRooms = NULL; -HashList *RSSFetchUrls = NULL; +void DeleteRssCfg(void *vptr) +{ + rss_aggregator *rncptr = (rss_aggregator *)vptr; + AsyncIO *IO = &rncptr->IO; + EVM_syslog(LOG_DEBUG, "RSS: destroying\n"); -/* - while (fgets(buf, sizeof buf, fp) != NULL && !CtdlThreadCheckStop()) { - buf[strlen(buf)-1] = 0; + FreeStrBuf(&rncptr->Url); + FreeStrBuf(&rncptr->rooms); + FreeStrBuf(&rncptr->CData); + FreeStrBuf(&rncptr->Key); + FreeStrBuf(&rncptr->IO.HttpReq.ReplyData); + DeleteHash(&rncptr->OtherQRnumbers); + FreeURL(&rncptr->IO.ConnectMe); - extract_token(instr, buf, 0, '|', sizeof instr); - if (!strcasecmp(instr, "rssclient")) { + DeleteHashPos (&rncptr->Pos); + DeleteHash (&rncptr->Messages); + if (rncptr->recp.recp_room != NULL) + free(rncptr->recp.recp_room); - use_this_rncptr = NULL; - extract_token(feedurl, buf, 1, '|', sizeof feedurl); + 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); +} - /* If any other rooms have requested the same feed, then we will just add this - * room to the target list for that client request. - * / TODO: how do we do this best? - for (rncptr=rnclist; rncptr!=NULL; rncptr=rncptr->next) { - if (!strcmp(ChrPtr(rncptr->Url), feedurl)) { - use_this_rncptr = rncptr; - } - } - * / - /* 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; - - rncptr->Url = NewStrBufPlain(feedurl, -1); - rncptr->rooms = NULL; - rnclist = rncptr; - use_this_rncptr = rncptr; +eNextState RSSAggregator_Terminate(AsyncIO *IO) +{ + rss_aggregator *rncptr = (rss_aggregator *)IO->Data; - } + EVM_syslog(LOG_DEBUG, "RSS: Terminating.\n"); - /* 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; - } - } - } - } - } - */ -typedef struct __RoomCounter { - int count; - long QRnumber; -} RoomCounter; + UnlinkRSSAggregator(rncptr); + return eAbort; +} +eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO) +{ + const char *pUrl; + rss_aggregator *rncptr = (rss_aggregator *)IO->Data; + pUrl = IO->ConnectMe->PlainUrl; + if (pUrl == NULL) + pUrl = ""; + EV_syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.\n", pUrl); -void DeleteRssCfg(void *vptr) -{ - rssnetcfg *rncptr = (rssnetcfg *)vptr; - FreeStrBuf(&rncptr->Url); - if (rncptr->rooms != NULL) free(rncptr->rooms); - free(rncptr); + 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; + StrBuf *CfgData=NULL; StrBuf *CfgType; StrBuf *Line; - RoomCounter *Count = NULL; + rss_room_counter *Count = NULL; struct stat statbuf; char filename[PATH_MAX]; - //char buf[1024]; - //char instr[32]; int fd; int Done; - //char feedurl[256]; - rssnetcfg *rncptr = NULL; - rssnetcfg *use_this_rncptr = NULL; - //int len = 0; - //char *ptr = NULL; + rss_aggregator *rncptr = NULL; + rss_aggregator *use_this_rncptr = NULL; void *vptr; const char *CfgPtr, *lPtr; const char *Err; - citthread_mutex_lock(&RSSQueueMutex); + pthread_mutex_lock(&RSSQueueMutex); if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr)) { - //CtdlLogPrintf(CTDL_DEBUG, "rssclient: %s already in progress.\n", qrbuf->QRname); - citthread_mutex_unlock(&RSSQueueMutex); + syslog(LOG_DEBUG, + "rssclient: [%ld] %s already in progress.\n", + qrbuf->QRnumber, + qrbuf->QRname); + pthread_mutex_unlock(&RSSQueueMutex); return; } - citthread_mutex_unlock(&RSSQueueMutex); + 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 */ fd = open(filename, 0); if (fd <= 0) { - //CtdlLogPrintf(CTDL_DEBUG, "rssclient: %s no config.\n", qrbuf->QRname); + /* syslog(LOG_DEBUG, + "rssclient: %s no config.\n", + qrbuf->QRname); */ return; } - if (CtdlThreadCheckStop()) + + if (server_shutting_down) return; + if (fstat(fd, &statbuf) == -1) { - CtdlLogPrintf(CTDL_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; } - if (CtdlThreadCheckStop()) + + if (server_shutting_down) return; + CfgData = NewStrBufPlain(NULL, statbuf.st_size + 1); + if (StrBufReadBLOB(CfgData, &fd, 1, statbuf.st_size, &Err) < 0) { close(fd); FreeStrBuf(&CfgData); - CtdlLogPrintf(CTDL_DEBUG, "ERROR: reading config '%s' - %s
\n", + syslog(LOG_DEBUG, "ERROR: reading config '%s' - %s
\n", filename, strerror(errno)); return; } close(fd); - if (CtdlThreadCheckStop()) + if (server_shutting_down) return; - + CfgPtr = NULL; CfgType = NewStrBuf(); Line = NewStrBufPlain(NULL, StrLength(CfgData)); @@ -518,56 +604,69 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) { lPtr = NULL; StrBufExtract_NextToken(CfgType, Line, &lPtr, '|'); - if (!strcmp("rssclient", ChrPtr(CfgType))) + if (!strcasecmp("rssclient", ChrPtr(CfgType))) { if (Count == NULL) { - Count = malloc(sizeof(RoomCounter)); + Count = malloc(sizeof(rss_room_counter)); Count->count = 0; } Count->count ++; - rncptr = (rssnetcfg *) malloc(sizeof(rssnetcfg)); - memset (rncptr, 0, sizeof(rssnetcfg)); + 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, '|'); - citthread_mutex_lock(&RSSQueueMutex); + pthread_mutex_lock(&RSSQueueMutex); GetHash(RSSFetchUrls, SKEY(rncptr->Url), &vptr); - use_this_rncptr = (rssnetcfg *)vptr; - citthread_mutex_unlock(&RSSQueueMutex); - + use_this_rncptr = (rss_aggregator *)vptr; if (use_this_rncptr != NULL) { - /* mustn't attach to an active session */ - if (use_this_rncptr->Attached == 1) - { - DeleteRssCfg(rncptr); - } - else - { - /* TODO: hook us into the otherone here. */ - } - - continue; + 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 = NULL; - rncptr->rooms = strdup(qrbuf->QRname); - citthread_mutex_lock(&RSSQueueMutex); + rncptr->rooms = NewStrBufPlain(qrbuf->QRname, -1); + + pthread_mutex_lock(&RSSQueueMutex); Put(RSSFetchUrls, SKEY(rncptr->Url), rncptr, DeleteRssCfg); - citthread_mutex_unlock(&RSSQueueMutex); + pthread_mutex_unlock(&RSSQueueMutex); } } } if (Count != NULL) { Count->QRnumber = qrbuf->QRnumber; - citthread_mutex_lock(&RSSQueueMutex); + pthread_mutex_lock(&RSSQueueMutex); + syslog(LOG_DEBUG, "rssclient: [%ld] %s now starting.\n", + qrbuf->QRnumber, qrbuf->QRname); Put(RSSQueueRooms, LKEY(qrbuf->QRnumber), Count, NULL); - citthread_mutex_unlock(&RSSQueueMutex); + pthread_mutex_unlock(&RSSQueueMutex); } FreeStrBuf(&CfgData); FreeStrBuf(&CfgType); @@ -579,43 +678,53 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) */ void rssclient_scan(void) { 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 - * 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; - CtdlLogPrintf(CTDL_DEBUG, "rssclient started\n"); + become_session(&rss_CC); + syslog(LOG_DEBUG, "rssclient started\n"); CtdlForEachRoom(rssclient_scan_room, NULL); - citthread_mutex_lock(&RSSQueueMutex); + pthread_mutex_lock(&RSSQueueMutex); - it = GetNewHashPos(RSSQueueRooms, 0); - while (GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && + it = GetNewHashPos(RSSFetchUrls, 0); + while (!server_shutting_down && + GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && (vrptr != NULL)) { - rptr = (rssnetcfg *)vrptr; - if (!rptr->Attached) rss_do_fetching(rptr); + rptr = (rss_aggregator *)vrptr; + if (!rss_do_fetching(rptr)) + UnlinkRSSAggregator(rptr); } DeleteHashPos(&it); - citthread_mutex_unlock(&RSSQueueMutex); + pthread_mutex_unlock(&RSSQueueMutex); - CtdlLogPrintf(CTDL_DEBUG, "rssclientscheduler ended\n"); + syslog(LOG_DEBUG, "rssclient ended\n"); doing_rssclient = 0; return; } -void RSSCleanup(void) +void rss_cleanup(void) { - citthread_mutex_destroy(&RSSQueueMutex); + /* citthread_mutex_destroy(&RSSQueueMutex); TODO */ DeleteHash(&RSSFetchUrls); DeleteHash(&RSSQueueRooms); } @@ -625,11 +734,13 @@ CTDL_MODULE_INIT(rssclient) { if (threading) { - citthread_mutex_init(&RSSQueueMutex, NULL); - RSSQueueRooms = NewHash(1, Flathash); + CtdlFillSystemContext(&rss_CC, "rssclient"); + pthread_mutex_init(&RSSQueueMutex, NULL); + RSSQueueRooms = NewHash(1, lFlathash); RSSFetchUrls = NewHash(1, NULL); - CtdlLogPrintf(CTDL_INFO, "%s\n", curl_version()); + syslog(LOG_INFO, "%s\n", curl_version()); CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER); + CtdlRegisterCleanupHook(rss_cleanup); } return "rssclient"; }