From e839bbab093a1a73198b3be9ad72f18cecb20f65 Mon Sep 17 00:00:00 2001 From: Wilfried Goesgens Date: Thu, 7 Jul 2011 22:37:22 +0000 Subject: [PATCH] work on making RSS aggregator instances and roomlists consistant. --- citadel/modules/rssclient/rss_atom_parser.c | 102 ++++----- citadel/modules/rssclient/rss_atom_parser.h | 68 +++--- citadel/modules/rssclient/serv_rssclient.c | 242 +++++++++++--------- citadel/utils/setup.c | 2 - 4 files changed, 219 insertions(+), 195 deletions(-) diff --git a/citadel/modules/rssclient/rss_atom_parser.c b/citadel/modules/rssclient/rss_atom_parser.c index 9bd7cf362..e86d30b84 100644 --- a/citadel/modules/rssclient/rss_atom_parser.c +++ b/citadel/modules/rssclient/rss_atom_parser.c @@ -135,8 +135,7 @@ void flush_rss_item(rss_item *ri) void rss_xml_start(void *data, const char *supplied_el, const char **attr) { rss_xml_handler *h; - rsscollection *rssc = (rsscollection*) data; - rssnetcfg *Cfg = rssc->Cfg; + rss_aggregator *rssc = (rss_aggregator*) data; rss_item *ri = rssc->Item; void *pv; const char *pel; @@ -173,19 +172,19 @@ void rss_xml_start(void *data, const char *supplied_el, const char **attr) rssc->Current = h = (rss_xml_handler*) pv; if (((h->Flags & RSS_UNSET) != 0) && - (Cfg->ItemType == RSS_UNSET)) + (rssc->ItemType == RSS_UNSET)) { - h->Handler(rssc->CData, ri, Cfg, attr); + h->Handler(rssc->CData, ri, rssc, attr); } else if (((h->Flags & RSS_RSS) != 0) && - (Cfg->ItemType == RSS_RSS)) + (rssc->ItemType == RSS_RSS)) { - h->Handler(rssc->CData, ri, Cfg, attr); + h->Handler(rssc->CData, ri, rssc, attr); } else if (((h->Flags & RSS_ATOM) != 0) && - (Cfg->ItemType == RSS_ATOM)) + (rssc->ItemType == RSS_ATOM)) { - h->Handler(rssc->CData, ri, Cfg, attr); + h->Handler(rssc->CData, ri, rssc, attr); } #ifdef DEBUG_RSS else @@ -201,8 +200,7 @@ void rss_xml_start(void *data, const char *supplied_el, const char **attr) void rss_xml_end(void *data, const char *supplied_el) { rss_xml_handler *h; - rsscollection *rssc = (rsscollection*) data; - rssnetcfg *Cfg = rssc->Cfg; + rss_aggregator *rssc = (rss_aggregator*) data; rss_item *ri = rssc->Item; const char *pel; char *sep = NULL; @@ -239,19 +237,19 @@ void rss_xml_end(void *data, const char *supplied_el) h = (rss_xml_handler*) pv; if (((h->Flags & RSS_UNSET) != 0) && - (Cfg->ItemType == RSS_UNSET)) + (rssc->ItemType == RSS_UNSET)) { - h->Handler(rssc->CData, ri, Cfg, NULL); + h->Handler(rssc->CData, ri, rssc, NULL); } else if (((h->Flags & RSS_RSS) != 0) && - (Cfg->ItemType == RSS_RSS)) + (rssc->ItemType == RSS_RSS)) { - h->Handler(rssc->CData, ri, Cfg, NULL); + h->Handler(rssc->CData, ri, rssc, NULL); } else if (((h->Flags & RSS_ATOM) != 0) && - (Cfg->ItemType == RSS_ATOM)) + (rssc->ItemType == RSS_ATOM)) { - h->Handler(rssc->CData, ri, Cfg, NULL); + h->Handler(rssc->CData, ri, rssc, NULL); } #ifdef DEBUG_RSS else @@ -270,39 +268,39 @@ void rss_xml_end(void *data, const char *supplied_el) -void RSS_item_rss_start (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_rss_start (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { CtdlLogPrintf(CTDL_DEBUG, "RSS: This is an RSS feed.\n"); Cfg->ItemType = RSS_RSS; } -void RSS_item_rdf_start(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_rdf_start(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { CtdlLogPrintf(CTDL_DEBUG, "RSS: This is an RDF feed.\n"); Cfg->ItemType = RSS_RSS; } -void ATOM_item_feed_start(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_feed_start(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { CtdlLogPrintf(CTDL_DEBUG, "RSS: This is an ATOM feed.\n"); Cfg->ItemType = RSS_ATOM; } -void RSS_item_item_start(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_item_start(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { ri->item_tag_nesting ++; flush_rss_item(ri); } -void ATOM_item_entry_start(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_entry_start(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { /* Atom feed... */ ri->item_tag_nesting ++; flush_rss_item(ri); } -void ATOM_item_link_start (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_link_start (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { int i; const char *pHref = NULL; @@ -374,7 +372,7 @@ void ATOM_item_link_start (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const ch -void ATOMRSS_item_title_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOMRSS_item_title_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if ((ri->item_tag_nesting == 0) && (StrLength(CData) > 0)) { NewStrBufDupAppendFlush(&ri->channel_title, CData, NULL, 0); @@ -382,14 +380,14 @@ void ATOMRSS_item_title_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const c } } -void RSS_item_guid_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_guid_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->guid, CData, NULL, 0); } } -void ATOM_item_id_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_id_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->guid, CData, NULL, 0); @@ -397,14 +395,14 @@ void ATOM_item_id_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** } -void RSS_item_link_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_link_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->link, CData, NULL, 0); StrBufTrim(ri->link); } } -void RSS_item_relink_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_relink_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->reLink, CData, NULL, 0); @@ -412,7 +410,7 @@ void RSS_item_relink_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char } } -void RSSATOM_item_title_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSSATOM_item_title_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->title, CData, NULL, 0); @@ -420,7 +418,7 @@ void RSSATOM_item_title_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const } } -void ATOM_item_content_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_content_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { long olen = StrLength (ri->description); long clen = StrLength (CData); @@ -437,7 +435,7 @@ void ATOM_item_content_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const c } } } -void ATOM_item_summary_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_summary_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { /* this can contain an abstract of the article. but we don't want to verwrite a full document if we already have it. */ if ((StrLength(CData) > 0) && (StrLength(ri->description) == 0)) @@ -447,7 +445,7 @@ void ATOM_item_summary_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const c } } -void RSS_item_description_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_description_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { long olen = StrLength (ri->description); long clen = StrLength (CData); @@ -465,7 +463,7 @@ void RSS_item_description_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, cons } } -void ATOM_item_published_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_published_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { StrBufTrim(CData); @@ -473,7 +471,7 @@ void ATOM_item_published_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const } } -void ATOM_item_updated_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_updated_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { StrBufTrim(CData); @@ -481,7 +479,7 @@ void ATOM_item_updated_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const c } } -void RSS_item_pubdate_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_pubdate_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { StrBufTrim(CData); @@ -490,7 +488,7 @@ void RSS_item_pubdate_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const ch } -void RSS_item_date_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_date_end (StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { StrBufTrim(CData); @@ -500,7 +498,7 @@ void RSS_item_date_end (StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char* -void RSS_item_author_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_author_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->author_or_creator, CData, NULL, 0); @@ -509,7 +507,7 @@ void RSS_item_author_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char } -void ATOM_item_name_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_name_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->author_or_creator, CData, NULL, 0); @@ -517,7 +515,7 @@ void ATOM_item_name_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char* } } -void ATOM_item_email_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_email_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->author_email, CData, NULL, 0); @@ -525,7 +523,7 @@ void ATOM_item_email_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char } } -void RSS_item_creator_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_creator_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if ((StrLength(CData) > 0) && (StrLength(ri->author_or_creator) == 0)) @@ -536,7 +534,7 @@ void RSS_item_creator_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const cha } -void ATOM_item_uri_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_uri_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { if (StrLength(CData) > 0) { NewStrBufDupAppendFlush(&ri->author_url, CData, NULL, 0); @@ -544,33 +542,33 @@ void ATOM_item_uri_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** } } -void RSS_item_item_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_item_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { --ri->item_tag_nesting; - rss_save_item(ri); + rss_save_item(ri, Cfg); } -void ATOM_item_entry_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void ATOM_item_entry_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { --ri->item_tag_nesting; - rss_save_item(ri); + rss_save_item(ri, Cfg); } -void RSS_item_rss_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_rss_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { // CtdlLogPrintf(CTDL_DEBUG, "End of feed detected. Closing parser.\n"); ri->done_parsing = 1; } -void RSS_item_rdf_end(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSS_item_rdf_end(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { // CtdlLogPrintf(CTDL_DEBUG, "End of feed detected. Closing parser.\n"); ri->done_parsing = 1; } -void RSSATOM_item_ignore(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char** Attr) +void RSSATOM_item_ignore(StrBuf *CData, rss_item *ri, rss_aggregator *Cfg, const char** Attr) { } @@ -581,7 +579,7 @@ void RSSATOM_item_ignore(StrBuf *CData, rss_item *ri, rssnetcfg *Cfg, const char */ void rss_xml_cdata_start(void *data) { - rsscollection *rssc = (rsscollection*) data; + rss_aggregator *rssc = (rss_aggregator*) data; FlushStrBuf(rssc->CData); } @@ -591,7 +589,7 @@ void rss_xml_cdata_end(void *data) } void rss_xml_chardata(void *data, const XML_Char *s, int len) { - rsscollection *rssc = (rsscollection*) data; + rss_aggregator *rssc = (rss_aggregator*) data; StrBufAppendBufPlain (rssc->CData, s, len, 0); } @@ -609,7 +607,7 @@ size_t rss_libcurl_callback(void *ptr, size_t size, size_t nmemb, void *stream) eNextState ParseRSSReply(AsyncIO *IO) { - rsscollection *rssc; + rss_aggregator *rssc; rss_item *ri; const char *at; char *ptr; @@ -617,8 +615,6 @@ eNextState ParseRSSReply(AsyncIO *IO) rssc = IO->Data; ri = rssc->Item; - ri->roomlist_parts = rssc->Cfg->roomlist_parts; - ri->roomlist = rssc->Cfg->rooms; rssc->CData = NewStrBufPlain(NULL, SIZ); rssc->Key = NewStrBuf(); at = NULL; @@ -641,7 +637,7 @@ eNextState ParseRSSReply(AsyncIO *IO) else ptr = "UTF-8"; - CtdlLogPrintf(CTDL_ALERT, "RSS: Now parsing [%s] \n", ChrPtr(rssc->Cfg->Url)); + CtdlLogPrintf(CTDL_ALERT, "RSS: Now parsing [%s] \n", ChrPtr(rssc->Url)); rssc->xp = XML_ParserCreateNS(ptr, ':'); if (!rssc->xp) { diff --git a/citadel/modules/rssclient/rss_atom_parser.h b/citadel/modules/rssclient/rss_atom_parser.h index e2f6e4d09..e6deb8a8a 100644 --- a/citadel/modules/rssclient/rss_atom_parser.h +++ b/citadel/modules/rssclient/rss_atom_parser.h @@ -24,12 +24,25 @@ #define RSS_ATOM (1<<2) #define RSS_REQUIRE_BUF (1<<3) -typedef struct _rss_item { +typedef struct rss_aggregator rss_aggregator; +typedef struct rss_item rss_item; +typedef struct rss_room_counter rss_room_counter; + +typedef void (*rss_handler_func)(StrBuf *CData, + rss_item *ri, + rss_aggregator *Cfg, + const char** Attr); + + +typedef struct __rss_xml_handler { + int Flags; + rss_handler_func Handler; +}rss_xml_handler; + +struct rss_item { int done_parsing; int item_tag_nesting; - int roomlist_parts; time_t pubdate; - StrBuf *roomlist; StrBuf *guid; StrBuf *title; StrBuf *link; @@ -41,43 +54,40 @@ typedef struct _rss_item { StrBuf *author_or_creator; StrBuf *author_url; StrBuf *author_email; -}rss_item; - -typedef struct __rssnetcfg { - int Attached; - int ItemType; - int roomlist_parts; - time_t last_error_when; - time_t next_poll; - StrBuf *Url; - StrBuf *rooms; -}rssnetcfg; +}; -typedef void (*rss_handler_func)(StrBuf *CData, - rss_item *ri, - rssnetcfg *Cfg, - const char** Attr); +struct rss_room_counter { + int count; + long QRnumber; +}; -typedef struct __rss_xml_handler { - int Flags; - rss_handler_func Handler; -}rss_xml_handler; - -typedef struct _rsscollection { +struct rss_aggregator { AsyncIO IO; XML_Parser xp; + + int RefCount; + int ItemType; + int roomlist_parts; + + time_t last_error_when; + time_t next_poll; + StrBuf *Url; + StrBuf *rooms; + long QRnumber; + HashList *OtherQRnumbers; - StrBuf *CData; - StrBuf *Key; + StrBuf *CData; + StrBuf *Key; rss_item *Item; - rssnetcfg *Cfg; rss_xml_handler *Current; -} rsscollection; +}; + + eNextState ParseRSSReply(AsyncIO *IO); -void rss_save_item(rss_item *ri); +void rss_save_item(rss_item *ri, rss_aggregator *Cfg); diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c index a51cac413..a21fcb3f3 100644 --- a/citadel/modules/rssclient/serv_rssclient.c +++ b/citadel/modules/rssclient/serv_rssclient.c @@ -62,6 +62,11 @@ #define TMP_SHORTER_URL_OFFSET 0xFE #define TMP_SHORTER_URLS 0xFD +citthread_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. */ + + struct rssnetcfg *rnclist = NULL; void AppendLink(StrBuf *Message, StrBuf *link, StrBuf *LinkTitle, const char *Title) @@ -84,15 +89,84 @@ typedef struct __networker_save_message { AsyncIO IO; struct CtdlMessage *Msg; struct recptypes *recp; + rss_aggregator *Cfg; StrBuf *MsgGUID; StrBuf *Message; struct UseTable ut; } networker_save_message; + +void DeleteRoomReference(long QRnumber) +{ + HashPos *At; + long HKLen; + const char *HK; + void *vData; + rss_room_counter *pRoomC; + + At = GetNewHashPos(RSSQueueRooms, 0); + + GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At); + GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData); + 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); + } + + DeleteHashPos(&At); + } + +} + +void UnlinkAggregator(rss_aggregator *Cfg) +{ + HashPos *At; + + UnlinkRooms(Cfg); + + At = GetNewHashPos(RSSFetchUrls, 0); + if (GetHashPosFromKey(RSSFetchUrls, SKEY(Cfg->Url), At) == 0) + { + DeleteEntryFromHash(RSSFetchUrls, At); + } + DeleteHashPos(&At); +} + eNextState FreeNetworkSaveMessage (AsyncIO *IO) { networker_save_message *Ctx = (networker_save_message *) IO->Data; + citthread_mutex_lock(&RSSQueueMutex); + Ctx->Cfg->RefCount --; + + if (Ctx->Cfg->RefCount == 0) + { + UnlinkAggregator(Ctx->Cfg); + + } + citthread_mutex_unlock(&RSSQueueMutex); + CtdlFreeMessage(Ctx->Msg); free_recipients(Ctx->recp); FreeStrBuf(&Ctx->MsgGUID); @@ -150,7 +224,7 @@ eNextState FetchNetworkUsetableEntry(AsyncIO *IO) return eSendMore; } } -void RSSQueueSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf *MsgGUID, StrBuf *MessageBody) +void RSSQueueSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf *MsgGUID, StrBuf *MessageBody, rss_aggregator *Cfg) { networker_save_message *Ctx; @@ -160,6 +234,7 @@ void RSSQueueSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf Ctx->MsgGUID = MsgGUID; Ctx->Message = MessageBody; Ctx->Msg = Msg; + Ctx->Cfg = Cfg; Ctx->recp = recp; Ctx->IO.Data = Ctx; Ctx->IO.CitContext = CloneContext(CC); @@ -172,7 +247,7 @@ void RSSQueueSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf /* * 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; @@ -187,11 +262,12 @@ void rss_save_item(rss_item *ri) recp = (struct recptypes *) malloc(sizeof(struct recptypes)); if (recp == NULL) return; memset(recp, 0, sizeof(struct recptypes)); - Buf = NewStrBufDup(ri->roomlist); + Buf = NewStrBufDup(Cfg->rooms); recp->recp_room = SmashStrBuf(&Buf); - recp->num_room = ri->roomlist_parts; + recp->num_room = Cfg->roomlist_parts; recp->recptypes_magic = RECPTYPES_MAGIC; + Cfg->RefCount ++; /* Construct a GUID to use in the S_USETABLE table. * If one is not present in the item itself, make one up. */ @@ -320,7 +396,7 @@ 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); + RSSQueueSaveMessage(msg, recp, guid, Message, Cfg); } @@ -328,8 +404,8 @@ 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; @@ -338,19 +414,15 @@ void rss_do_fetching(rssnetcfg *Cfg) { now = time(NULL); if ((Cfg->next_poll != 0) && (now < Cfg->next_poll)) - return; - Cfg->Attached = 1; + return 0; + Cfg->RefCount = 1; 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; + Cfg->Item = ri; + IO = &Cfg->IO; IO->CitContext = CloneContext(CC); - IO->Data = rssc; - ri->roomlist = Cfg->rooms; + IO->Data = Cfg; CtdlLogPrintf(CTDL_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(Cfg->Url)); @@ -364,79 +436,18 @@ void rss_do_fetching(rssnetcfg *Cfg) { ParseRSSReply)) { CtdlLogPrintf(CTDL_ALERT, "Unable to initialize libcurl.\n"); -// goto abort; + return 0; } evcurl_handle_start(IO); + return 1; } -citthread_mutex_t RSSQueueMutex; /* locks the access to the following vars: */ -HashList *RSSQueueRooms = NULL; -HashList *RSSFetchUrls = NULL; - - -/* - 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; - - extract_token(feedurl, buf, 1, '|', sizeof feedurl); - - /* 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; - - } - - /* 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; - void DeleteRssCfg(void *vptr) { - rssnetcfg *rncptr = (rssnetcfg *)vptr; + rss_aggregator *rncptr = (rss_aggregator *)vptr; FreeStrBuf(&rncptr->Url); FreeStrBuf(&rncptr->rooms); @@ -452,18 +463,13 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) StrBuf *CfgData; 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; @@ -524,38 +530,48 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data) { 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_room_counter)); rncptr->roomlist_parts = 1; rncptr->Url = NewStrBuf(); StrBufExtract_NextToken(rncptr->Url, Line, &lPtr, '|'); citthread_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 - { - StrBufAppendBufPlain(use_this_rncptr->rooms, - qrbuf->QRname, - -1, 0); - use_this_rncptr->roomlist_parts++; - } - - continue; + /* mustn't attach to an active session */ + if (use_this_rncptr->RefCount > 0) + { + DeleteRssCfg(rncptr); + Count->count--; + } + else + { + 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); + +//// TODO add reference here! + } + QRnumber = (long*)malloc(sizeof(long)); + *QRnumber = qrbuf->QRnumber; + Put(use_this_rncptr->OtherQRnumbers, LKEY(qrbuf->QRnumber), QRnumber, NULL); + use_this_rncptr->roomlist_parts++; + } + citthread_mutex_unlock(&RSSQueueMutex); + continue; } + citthread_mutex_unlock(&RSSQueueMutex); rncptr->ItemType = RSS_UNSET; @@ -584,7 +600,7 @@ 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; @@ -607,8 +623,12 @@ void rssclient_scan(void) { it = GetNewHashPos(RSSQueueRooms, 0); while (GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && (vrptr != NULL)) { - rptr = (rssnetcfg *)vrptr; - if (!rptr->Attached) rss_do_fetching(rptr); + rptr = (rss_aggregator *)vrptr; + if (rptr->RefCount == 0) + if (!rss_do_fetching(rptr)) + { + /// TODO: flush me. + } } DeleteHashPos(&it); citthread_mutex_unlock(&RSSQueueMutex); diff --git a/citadel/utils/setup.c b/citadel/utils/setup.c index da770f7d1..c6d907a2b 100644 --- a/citadel/utils/setup.c +++ b/citadel/utils/setup.c @@ -261,8 +261,6 @@ void cit_backtrace(void) struct config config; - -struct config config; int direction; -- 2.30.2