]> code.citadel.org Git - citadel.git/blobdiff - citadel/modules/rssclient/serv_rssclient.c
Learned that LOG_ALERT is not really appropriate for the kind of things
[citadel.git] / citadel / modules / rssclient / serv_rssclient.c
index 906e2787c60eba39dc9a3616faacfd627f945816..c2dfa85e1ad1149aff91919df492ff68edade82a 100644 (file)
@@ -1,8 +1,15 @@
 /*
- * $Id: serv_rssclient.c 5652 2007-10-29 20:14:48Z ajc $
- *
  * Bring external RSS feeds into rooms.
  *
+ * Copyright (c) 2007-2016 by the citadel.org team
+ *
+ * This program is open source software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 3.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.See the
+ * GNU General Public License for more details.
  */
 
 #include <stdlib.h>
@@ -14,9 +21,9 @@
 # include <time.h>
 #else
 # if HAVE_SYS_TIME_H
-#  include <sys/time.h>
+#include <sys/time.h>
 # else
-#  include <time.h>
+#include <time.h>
 # endif
 #endif
 
 #include <errno.h>
 #include <sys/types.h>
 #include <sys/stat.h>
+#include <expat.h>
+#include <curl/curl.h>
+#include <libcitadel.h>
 #include "citadel.h"
 #include "server.h"
 #include "citserver.h"
 #include "support.h"
 #include "config.h"
-#include "tools.h"
-#include "room_ops.h"
+#include "threads.h"
 #include "ctdl_module.h"
-#include "clientsocket.h"
 #include "msgbase.h"
+#include "parsedate.h"
 #include "database.h"
 #include "citadel_dirs.h"
 #include "md5.h"
+#include "context.h"
+#include "event_client.h"
+#include "rss_atom_parser.h"
 
-#ifdef HAVE_EXPAT
-#include <expat.h>
 
+#define TMP_MSGDATA 0xFF
+#define TMP_SHORTER_URL_OFFSET 0xFE
+#define TMP_SHORTER_URLS 0xFD
 
-struct rssnetcfg {
-       struct rssnetcfg *next;
-       char url[256];
-       char *rooms;
-};
+time_t last_run = 0L;
 
-struct rss_item {
-       char *chardata;
-       int chardata_len;
-       char *roomlist;
-       int done_parsing;
-       char *guid;
-       char *title;
-       char *link;
-       char *description;
-       time_t pubdate;
-       char channel_title[256];
-       int item_tag_nesting;
-};
-
-struct rssnetcfg *rnclist = NULL;
+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_TerminateDB(AsyncIO *IO);
+eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO);
+struct CitContext rss_CC;
 
-/*
- * Commit a fetched and parsed RSS item to disk
- */
-void rss_save_item(struct rss_item *ri) {
+struct rssnetcfg *rnclist = NULL;
+int RSSClientDebugEnabled = 0;
+#define N ((rss_aggregator*)IO->Data)->Cfg.QRnumber
+
+#define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (RSSClientDebugEnabled != 0))
+
+typedef enum _RSSState {
+       eRSSCreated,
+       eRSSFetching,
+       eRSSFailure,
+       eRSSParsing,
+       eRSSUT
+} RSSState;
+ConstStr RSSStates[] = {
+       {HKEY("Aggregator created")},
+       {HKEY("Fetching content")},
+       {HKEY("Failed")},
+       {HKEY("parsing content")},
+       {HKEY("checking usetable")}
+};
 
-       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;
 
-       recp = (struct recptypes *) malloc(sizeof(struct recptypes));
-       if (recp == NULL) return;
-       memset(recp, 0, sizeof(struct recptypes));
-       recp->recp_room = strdup(ri->roomlist);
-       recp->num_room = num_tokens(ri->roomlist, '|');
-       recp->recptypes_magic = RECPTYPES_MAGIC;
-   
-       /* Construct a GUID to use in the S_USETABLE table.
-        * If one is not present in the item itself, make one up.
-        */
-       if (ri->guid != NULL) {
-               snprintf(utmsgid, sizeof utmsgid, "rss/%s", ri->guid);
-       }
-       else {
-               MD5Init(&md5context);
-               if (ri->title != NULL) {
-                       MD5Update(&md5context, ri->title, strlen(ri->title));
-               }
-               if (ri->link != NULL) {
-                       MD5Update(&md5context, ri->link, strlen(ri->link));
+static size_t GetLocationString( void *ptr, size_t size, size_t nmemb, void *userdata)
+{
+#define LOCATION "location"
+       if (strncasecmp((char*)ptr, LOCATION, sizeof(LOCATION) - 1) == 0)
+       {
+               AsyncIO *IO = (AsyncIO *) userdata;
+               rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
+
+               char *pch = (char*) ptr;
+               char *pche;
+               
+               pche = pch + (size * nmemb);
+               pch += sizeof(LOCATION);
+               
+               while (isspace(*pch) || (*pch == ':'))
+                       pch ++;
+
+               while (isspace(*pche) || (*pche == '\0'))
+                       pche--;
+               if (RSSAggr->RedirectUrl == NULL) {
+                       RSSAggr->RedirectUrl = NewStrBufPlain(pch, pche - pch + 1);
                }
-               MD5Final(rawdigest, &md5context);
-               for (i=0; i<MD5_DIGEST_LEN; i++) {
-                       sprintf(&utmsgid[i*2], "%02X", (unsigned char) (rawdigest[i] & 0xff));
-                       utmsgid[i*2] = tolower(utmsgid[i*2]);
-                       utmsgid[(i*2)+1] = tolower(utmsgid[(i*2)+1]);
+               else {
+                       FlushStrBuf(RSSAggr->RedirectUrl);
+                       StrBufPlain(RSSAggr->RedirectUrl, pch, pche - pch + 1); 
                }
-               strcat(utmsgid, "_rss2ctdl");
        }
+       return size * nmemb;
+}
 
-       /* Find out if we've already seen this item */
-       cdbut = cdb_fetch(CDB_USETABLE, utmsgid, strlen(utmsgid));
-       if (cdbut != NULL) {
-               /* Item has already been seen */
-               lprintf(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) );
+static void SetRSSState(AsyncIO *IO, RSSState State)
+{
+       CitContext* CCC = IO->CitContext;
+       if (CCC != NULL) {
+               memcpy(CCC->cs_clientname, RSSStates[State].Key, RSSStates[State].len + 1);
        }
-       else {
-               /* Item has not been seen, so save it. */
+}
 
-               for (i=strlen(ri->description); i>=0; --i) {
-                       if (isspace(ri->description[i])) {
-                               ri->description[i] = ' ';
-                       }
+
+void DeleteRoomReference(long QRnumber)
+{
+       HashPos *At;
+       long HKLen;
+       const char *HK;
+       void *vData = NULL;
+       rss_room_counter *pRoomC;
+
+       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);
+}
 
-               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;
-               msg->cm_fields['A'] = strdup("rss");
-               msg->cm_fields['N'] = strdup(NODENAME);
-               msg->cm_fields['U'] = strdup(ri->title);
-               msg->cm_fields['T'] = malloc(64);
-               snprintf(msg->cm_fields['T'], 64, "%ld", ri->pubdate);
-               if (!IsEmptyStr(ri->channel_title)) {
-                       msg->cm_fields['O'] = strdup(ri->channel_title);
+void UnlinkRooms(rss_aggregator *RSSAggr)
+{
+       DeleteRoomReference(RSSAggr->Cfg.QRnumber);
+       if (RSSAggr->OtherQRnumbers != NULL)
+       {
+               long HKLen;
+               const char *HK;
+               HashPos *At;
+               void *vData;
+
+               At = GetNewHashPos(RSSAggr->OtherQRnumbers, 0);
+               while (! server_shutting_down &&
+                      GetNextHashPos(RSSAggr->OtherQRnumbers,
+                                     At,
+                                     &HKLen, &HK,
+                                     &vData) &&
+                      (vData != NULL))
+               {
+                       pRSSConfig *Data = (pRSSConfig*) vData;
+                       DeleteRoomReference(Data->QRnumber);
                }
 
-               msglen = 1024 + strlen(ri->link) + strlen(ri->description) ;
-               msg->cm_fields['M'] = malloc(msglen);
-               snprintf(msg->cm_fields['M'], msglen,
-                       "Content-type: text/html\r\n\r\n"
-                       "<html><body>\n"
-                       "%s<br><br>\n"
-                       "<a href=\"%s\">%s</a>\n"
-                       "</body></html>\n"
-                       ,
-                       ri->description,
-                       ri->link, ri->link
-               );
+               DeleteHashPos(&At);
+       }
+}
 
-               CtdlSubmitMsg(msg, recp, NULL);
-               CtdlFreeMessage(msg);
-               free_recipients(recp);
+void UnlinkRSSAggregator(rss_aggregator *RSSAggr)
+{
+       HashPos *At;
 
-               /* write the uidl to the use table so we don't store this item again */
-               strcpy(ut.ut_msgid, utmsgid);
-               ut.ut_timestamp = time(NULL);
-               cdb_store(CDB_USETABLE, utmsgid, strlen(utmsgid), &ut, sizeof(struct UseTable) );
+       pthread_mutex_lock(&RSSQueueMutex);
+       UnlinkRooms(RSSAggr);
+
+       At = GetNewHashPos(RSSFetchUrls, 0);
+       if (GetHashPosFromKey(RSSFetchUrls, SKEY(RSSAggr->Url), At))
+       {
+               DeleteEntryFromHash(RSSFetchUrls, At);
        }
+       DeleteHashPos(&At);
+       last_run = time(NULL);
+       pthread_mutex_unlock(&RSSQueueMutex);
 }
 
+void DeleteRssCfg(void *vptr)
+{
+       rss_aggregator *RSSAggr = (rss_aggregator *)vptr;
+       AsyncIO *IO = &RSSAggr->IO;
 
+       if (IO->CitContext != NULL) {
+               syslog(LOG_DEBUG, "RSS: destroying\n");
+       }
 
-/*
- * Convert an RDF/RSS datestamp into a time_t
- */
-time_t rdf_parsedate(char *p)
-{
-       struct tm tm;
+       FreeStrBuf(&RSSAggr->Url);
+       FreeStrBuf(&RSSAggr->RedirectUrl);
+       FreeStrBuf(&RSSAggr->rooms);
+       FreeStrBuf(&RSSAggr->CData);
+       FreeStrBuf(&RSSAggr->Key);
+       DeleteHash(&RSSAggr->OtherQRnumbers);
 
-       if (!p) return 0L;
-       if (strlen(p) < 10) return 0L;
+       DeleteHashPos (&RSSAggr->Pos);
+       DeleteHash (&RSSAggr->Messages);
+       if (RSSAggr->recp.recp_room != NULL)
+               free(RSSAggr->recp.recp_room);
 
-       memset(&tm, 0, sizeof tm);
 
-       /* YYYY-MM-DDTHH:MM format...
-        */
-       if ( (p[4] == '-') && (p[7] == '-') ) {
-               tm.tm_year = atoi(&p[0]) - 1900;
-               tm.tm_mon = atoi(&p[5]) - 1;
-               tm.tm_mday = atoi(&p[8]);
-               if ( (p[10] == 'T') && (p[13] == ':') ) {
-                       tm.tm_hour = atoi(&p[11]);
-                       tm.tm_min = atoi(&p[14]);
-               }
-       }
+       if (RSSAggr->Item != NULL)
+       {
+               flush_rss_item(RSSAggr->Item);
 
-       else {
-               /* FIXME try an imap timestamp conversion */
+               free(RSSAggr->Item);
        }
 
-       return mktime(&tm);
+       FreeAsyncIOContents(&RSSAggr->IO);
+       memset(RSSAggr, 0, sizeof(rss_aggregator));
+       free(RSSAggr);
 }
 
+eNextState RSSAggregator_Terminate(AsyncIO *IO)
+{
+       rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
 
+       syslog(LOG_DEBUG, "RSS: Terminating.");
 
-void rss_xml_start(void *data, const char *supplied_el, const char **attr) {
-       struct rss_item *ri = (struct rss_item *) data;
-       char el[256];
-       char *sep = NULL;
-
-       /* Axe the namespace, we don't care about it */
-       safestrncpy(el, supplied_el, sizeof el);
-       while (sep = strchr(el, ':'), sep) {
-               strcpy(el, ++sep);
-       }
+       StopCurlWatchers(IO);
+       UnlinkRSSAggregator(RSSAggr);
+       return eAbort;
+}
 
-       if (!strcasecmp(el, "item")) {
-               ++ri->item_tag_nesting;
-
-               /* Initialize the feed item data structure */
-               if (ri->guid != NULL) free(ri->guid);
-               ri->guid = NULL;
-               if (ri->title != NULL) free(ri->title);
-               ri->title = NULL;
-               if (ri->link != NULL) free(ri->link);
-               ri->link = NULL;
-               if (ri->description != NULL) free(ri->description);
-               ri->description = NULL;
-
-               /* Throw away any existing character data */
-               if (ri->chardata_len > 0) {
-                       free(ri->chardata);
-                       ri->chardata = 0;
-                       ri->chardata_len = 0;
-               }
-       }
+eNextState RSSAggregator_TerminateDB(AsyncIO *IO)
+{
+       rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
 
+       syslog(LOG_DEBUG, "RSS: Terminating.");
 
 
+       StopDBWatchers(&RSSAggr->IO);
+       UnlinkRSSAggregator(RSSAggr);
+       return eAbort;
 }
 
-void rss_xml_end(void *data, const char *supplied_el) {
-       struct rss_item *ri = (struct rss_item *) data;
-       char el[256];
-       char *sep = NULL;
+eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO)
+{
+       const char *pUrl;
+       rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
 
-       /* Axe the namespace, we don't care about it */
-       safestrncpy(el, supplied_el, sizeof el);
-       while (sep = strchr(el, ':'), sep) {
-               strcpy(el, ++sep);
-       }
+       pUrl = IO->ConnectMe->PlainUrl;
+       if (pUrl == NULL)
+               pUrl = "";
 
-       if ( (!strcasecmp(el, "title")) && (ri->item_tag_nesting == 0) ) {
-               safestrncpy(ri->channel_title, ri->chardata, sizeof ri->channel_title);
-               striplt(ri->channel_title);
-       }
+       syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.", pUrl);
 
-       if (!strcasecmp(el, "guid")) {
-               if (ri->guid != NULL) free(ri->guid);
-               striplt(ri->chardata);
-               ri->guid = strdup(ri->chardata);
-       }
+       StopCurlWatchers(IO);
+       UnlinkRSSAggregator(RSSAggr);
+       return eAbort;
+}
 
-       if (!strcasecmp(el, "title")) {
-               if (ri->title != NULL) free(ri->title);
-               striplt(ri->chardata);
-               ri->title = strdup(ri->chardata);
+void AppendLink(StrBuf *Message,
+               StrBuf *link,
+               StrBuf *LinkTitle,
+               const char *Title)
+{
+       if (StrLength(link) > 0)
+       {
+               StrBufAppendBufPlain(Message, HKEY("<a href=\""), 0);
+               StrBufAppendBuf(Message, 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("</a><br>\n"), 0);
        }
+}
 
-       if (!strcasecmp(el, "link")) {
-               if (ri->link != NULL) free(ri->link);
-               striplt(ri->chardata);
-               ri->link = strdup(ri->chardata);
-       }
 
-       if (!strcasecmp(el, "description")) {
-               if (ri->description != NULL) free(ri->description);
-               ri->description = strdup(ri->chardata);
-       }
+int rss_format_item(AsyncIO *IO, networker_save_message *SaveMsg)
+{
+       StrBuf *Message;
+       int msglen = 0;
 
-       if ( (!strcasecmp(el, "pubdate")) || (!strcasecmp(el, "date")) ) {
-               striplt(ri->chardata);
-               ri->pubdate = rdf_parsedate(ri->chardata);
-       }
+       if (StrLength(SaveMsg->description) + 
+           StrLength(SaveMsg->link) + 
+           StrLength(SaveMsg->linkTitle) + 
+           StrLength(SaveMsg->reLink) +
+           StrLength(SaveMsg->reLinkTitle) +
+           StrLength(SaveMsg->title) == 0)
+       {
+               syslog(LOG_INFO, "Refusing to save empty message.");
+               return 0;
+       }
+
+       CM_Flush(&SaveMsg->Msg);
+
+       if (SaveMsg->author_or_creator != NULL) {
+
+               char *From;
+               StrBuf *Encoded = NULL;
+               int FromAt;
+
+               From = html_to_ascii(ChrPtr(SaveMsg->author_or_creator),
+                                    StrLength(SaveMsg->author_or_creator),
+                                    512, 0);
+               StrBufPlain(SaveMsg->author_or_creator, From, -1);
+               StrBufTrim(SaveMsg->author_or_creator);
+               free(From);
+
+               FromAt = strchr(ChrPtr(SaveMsg->author_or_creator), '@') != NULL;
+               if (!FromAt && StrLength (SaveMsg->author_email) > 0)
+               {
+                       StrBufRFC2047encode(&Encoded, SaveMsg->author_or_creator);
+                       CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &Encoded);
+                       CM_SetAsFieldSB(&SaveMsg->Msg, eMessagePath, &SaveMsg->author_email);
+               }
+               else
+               {
+                       if (FromAt)
+                       {
+                               CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &SaveMsg->author_or_creator);
+                               CM_CopyField(&SaveMsg->Msg, eMessagePath, eAuthor);
+                       }
+                       else
+                       {
+                               StrBufRFC2047encode(&Encoded,
+                                                   SaveMsg->author_or_creator);
+                               CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &Encoded);
+                               CM_SetField(&SaveMsg->Msg, eMessagePath, HKEY("rss@localhost"));
 
-       if (!strcasecmp(el, "item")) {
-               --ri->item_tag_nesting;
-               rss_save_item(ri);
+                       }
+               }
        }
-
-       if ( (!strcasecmp(el, "rss")) || (!strcasecmp(el, "rdf")) ) {
-               lprintf(CTDL_DEBUG, "End of feed detected.  Closing parser.\n");
-               ri->done_parsing = 1;
+       else {
+               CM_SetField(&SaveMsg->Msg, eAuthor, HKEY("rss"));
+       }
+
+       CM_SetField(&SaveMsg->Msg, eNodeName, CtdlGetConfigStr("c_nodename"), strlen(CtdlGetConfigStr("c_nodename")));
+       if (SaveMsg->title != NULL) {
+               long len;
+               char *Sbj;
+               StrBuf *Encoded, *QPEncoded;
+
+               QPEncoded = NULL;
+               StrBufSpaceToBlank(SaveMsg->title);
+               len = StrLength(SaveMsg->title);
+               Sbj = html_to_ascii(ChrPtr(SaveMsg->title), len, 512, 0);
+               if (!IsEmptyStr(Sbj)) {
+                       len = strlen(Sbj);
+                       if ((Sbj[len - 1] == '\n'))
+                       {
+                               len --;
+                               Sbj[len] = '\0';
+                       }
+                       Encoded = NewStrBufPlain(Sbj, len);
+               
+
+                       StrBufTrim(Encoded);
+                       StrBufRFC2047encode(&QPEncoded, Encoded);
+                       
+                       CM_SetAsFieldSB(&SaveMsg->Msg, eMsgSubject, &QPEncoded);
+                       FreeStrBuf(&Encoded);
+               }
+               if (Sbj != NULL) {
+                       free(Sbj);
+               }
        }
+       if (SaveMsg->link == NULL)
+               SaveMsg->link = NewStrBufPlain(HKEY(""));
 
-       if (ri->chardata_len > 0) {
-               free(ri->chardata);
-               ri->chardata = 0;
-               ri->chardata_len = 0;
-       }
+#if 0 /* temporarily disable shorter urls. */
+       SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] =
+               GetShorterUrls(SaveMsg->description);
+#endif
 
-}
+       msglen += 1024 + StrLength(SaveMsg->link) + StrLength(SaveMsg->description) ;
 
+       Message = NewStrBufPlain(NULL, msglen);
 
-/*
- * This callback stores up the data which appears in between tags.
- */
-void rss_xml_chardata(void *data, const XML_Char *s, int len) {
-       struct rss_item *ri = (struct rss_item *) data;
-       int old_len;
-       int new_len;
-       char *new_buffer;
-
-       old_len = ri->chardata_len;
-       new_len = old_len + len;
-       new_buffer = realloc(ri->chardata, new_len + 1);
-       if (new_buffer != NULL) {
-               memcpy(&new_buffer[old_len], s, len);
-               new_buffer[new_len] = 0;
-               ri->chardata = new_buffer;
-               ri->chardata_len = new_len;
-       }
-}
+       StrBufPlain(Message, HKEY(
+                           "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n"
+                           "<html><body>\n"));
+#if 0 /* disable shorter url for now. */
+       SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message);
+#endif
+       StrBufAppendBuf(Message, SaveMsg->description, 0);
+       StrBufAppendBufPlain(Message, HKEY("<br><br>\n"), 0);
 
+       AppendLink(Message, SaveMsg->link, SaveMsg->linkTitle, NULL);
+       AppendLink(Message, SaveMsg->reLink, SaveMsg->reLinkTitle, "Reply to this");
+       StrBufAppendBufPlain(Message, HKEY("</body></html>\n"), 0);
 
+       SaveMsg->Message = Message;
+       return 1;
+}
 
-/* 
- * Parse a URL into host, port number, and resource identifier.
- */
-int parse_url(char *url, char *hostname, int *port, char *identifier)
+eNextState RSSSaveMessage(AsyncIO *IO)
 {
-       char protocol[1024];
-       char scratch[1024];
-       char *ptr = NULL;
-       char *nptr = NULL;
-       
-       strcpy(scratch, url);
-       ptr = (char *)strchr(scratch, ':');
-       if (!ptr) {
-               return(1);      /* no protocol specified */
-       }
+       long len;
+       const char *Key;
+       rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data;
 
-       strcpy(ptr, "");
-       strcpy(protocol, scratch);
-       if (strcmp(protocol, "http")) {
-               return(2);      /* not HTTP */
-       }
+       if (rss_format_item(IO, RSSAggr->ThisMsg))
+       {
+               CM_SetAsFieldSB(&RSSAggr->ThisMsg->Msg, eMesageText,
+                                      &RSSAggr->ThisMsg->Message);
 
-       strcpy(scratch, url);
-       ptr = (char *) strstr(scratch, "//");
-       if (!ptr) {
-               return(3);      /* no server specified */
+               CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0);
+               
+               /* write the uidl to the use table so we don't store this item again */
+               
+               CheckIfAlreadySeen("RSS Item Insert", RSSAggr->ThisMsg->MsgGUID, EvGetNow(IO), 0, eWrite, CCID, IO->ID);
        }
-       ptr += 2;
 
-       strcpy(hostname, ptr);
-       nptr = (char *)strchr(ptr, ':');
-       if (!nptr) {
-               *port = 80;     /* default */
-               nptr = (char *)strchr(hostname, '/');
-       }
-       else {
-               sscanf(nptr, ":%d", port);
-               nptr = (char *)strchr(hostname, ':');
-       }
+       if (GetNextHashPos(RSSAggr->Messages,
+                          RSSAggr->Pos,
+                          &len, &Key,
+                          (void**) &RSSAggr->ThisMsg))
+               return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
+       else
+               return eAbort;
+}
+
+eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
+{
+       static const time_t antiExpire = USETABLE_ANTIEXPIRE_HIRES;
+#ifndef DEBUG_RSS
+       time_t seenstamp = 0;
+       const char *Key;
+       long len;
+       rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
 
-       if (nptr) {
-               *nptr = '\0';
+       /* Find out if we've already seen this item */
+// todo: expiry?
+       SetRSSState(IO, eRSSUT);
+       seenstamp = CheckIfAlreadySeen("RSS Item Seen",
+                                      Ctx->ThisMsg->MsgGUID,
+                                      EvGetNow(IO),
+                                      antiExpire,
+                                      eCheckUpdate,
+                                      CCID, IO->ID);
+       if (seenstamp != 0)
+       {
+               /* Item has already been seen */
+               syslog(LOG_DEBUG, "%s has already been seen - %ld < %ld", ChrPtr(Ctx->ThisMsg->MsgGUID), seenstamp, antiExpire);
+
+               SetRSSState(IO, eRSSParsing);
+
+               if (GetNextHashPos(Ctx->Messages,
+                                  Ctx->Pos,
+                                  &len, &Key,
+                                  (void**) &Ctx->ThisMsg))
+                       return NextDBOperation(
+                               IO,
+                               RSS_FetchNetworkUsetableEntry);
+               else
+                       return eAbort;
        }
+       else
+#endif
+       {
+               /* Item has already been seen */
+               syslog(LOG_DEBUG,
+                             "%s Parsing - %ld >= %ld",
+                             ChrPtr(Ctx->ThisMsg->MsgGUID),
+                             seenstamp, antiExpire);
+               SetRSSState(IO, eRSSParsing);
 
-       nptr = (char *)strchr(ptr, '/');
-       
-       if (!nptr) {
-               return(4);      /* no url specified */
+               NextDBOperation(IO, RSSSaveMessage);
+               return eSendMore;
        }
-       
-       strcpy(identifier, nptr);
-       return(0);
+       return eSendMore;
 }
 
-
-/*
- * Begin a feed parse
- */
-void rss_do_fetching(char *url, char *rooms) {
-       char buf[1024];
-       char rsshost[1024];
-       int rssport = 80;
-       char rssurl[1024];
-       struct rss_item ri;
-       XML_Parser xp;
-       int sock = (-1);
-       int got_bytes = (-1);
-       int redirect_count = 0;
-
-       /* Parse the URL */
-       if (parse_url(url, rsshost, &rssport, rssurl) != 0) {
-               lprintf(CTDL_ALERT, "Invalid URL: %s\n", url);
+void UpdateLastKnownGood(pRSSConfig *pCfg, time_t now)
+{
+       OneRoomNetCfg *pRNCfg;
+       begin_critical_section(S_NETCONFIGS);
+       pRNCfg = CtdlGetNetCfgForRoom(pCfg->QRnumber);
+       if (pRNCfg != NULL)
+       {
+               RSSCfgLine *RSSCfg = (RSSCfgLine *)pRNCfg->NetConfigs[rssclient];
+
+               while (RSSCfg != NULL)
+               {
+                       if (RSSCfg == pCfg->pCfg)
+                               break;
+
+                       RSSCfg = RSSCfg->next;
+               }
+               if (RSSCfg != NULL)
+               {
+                       RSSCfg->last_known_good = now;
+               }
        }
+       SaveRoomNetConfigFile(pRNCfg, pCfg->QRnumber);
+       FreeRoomNetworkStruct(&pRNCfg);
+       end_critical_section(S_NETCONFIGS);
+}
 
-       xp = XML_ParserCreateNS("UTF-8", ':');
-       if (!xp) {
-               lprintf(CTDL_ALERT, "Cannot create XML parser!\n");
-               return;
-       }
+eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
+{
+       HashPos *it = NULL;
+       long len;
+       const char *Key;
+       pRSSConfig *pCfg;
+       u_char rawdigest[MD5_DIGEST_LEN];
+       struct MD5Context md5context;
+       StrBuf *guid;
+       rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
 
-       memset(&ri, 0, sizeof(struct rss_item));
-       ri.roomlist = rooms;
-       XML_SetElementHandler(xp, rss_xml_start, rss_xml_end);
-       XML_SetCharacterDataHandler(xp, rss_xml_chardata);
-       XML_SetUserData(xp, &ri);
-
-retry: lprintf(CTDL_NOTICE, "Connecting to <%s>\n", rsshost);
-       sprintf(buf, "%d", rssport);
-       sock = sock_connect(rsshost, buf, "tcp");
-       if (sock >= 0) {
-               lprintf(CTDL_DEBUG, "Connected!\n");
-
-               snprintf(buf, sizeof buf, "GET %s HTTP/1.0", rssurl);
-               lprintf(CTDL_DEBUG, "<%s\n", buf);
-               sock_puts(sock, buf);
-
-               snprintf(buf, sizeof buf, "Host: %s", rsshost);
-               lprintf(CTDL_DEBUG, "<%s\n", buf);
-               sock_puts(sock, buf);
-
-               sock_puts(sock, "");
-
-               if (sock_getln(sock, buf, sizeof buf) >= 0) {
-               lprintf(CTDL_DEBUG, ">%s\n", buf);
-                       remove_token(buf, 0, ' ');
-
-                       /* 200 OK */
-                       if (buf[0] == '2') {
-
-                               while (got_bytes = sock_getln(sock, buf, sizeof buf),
-                                     (got_bytes >= 0 && (strcmp(buf, "")) && (strcmp(buf, "\r"))) ) {
-                                       /* discard headers */
-                               }
-
-                               while (got_bytes = sock_read(sock, buf, sizeof buf, 0),
-                                     ((got_bytes>=0) && (ri.done_parsing == 0)) ) {
-                                       XML_Parse(xp, buf, got_bytes, 0);
-                               }
-                               if (ri.done_parsing == 0) XML_Parse(xp, "", 0, 1);
-                       }
 
-                       /* 30X redirect */
-                       else if ( (!strncmp(buf, "30", 2)) && (redirect_count < 16) ) {
-                               while (got_bytes = sock_getln(sock, buf, sizeof buf),
-                                     (got_bytes >= 0 && (strcmp(buf, "")) && (strcmp(buf, "\r"))) ) {
-                                       if (!strncasecmp(buf, "Location:", 9)) {
-                                               ++redirect_count;
-                                               strcpy(buf, &buf[9]);
-                                               striplt(buf);
-                                               if (parse_url(buf, rsshost, &rssport, rssurl) == 0) {
-                                                       goto retry;
-                                               }
-                                               else {
-                                                       lprintf(CTDL_ALERT, "Invalid URL: %s\n", buf);
-                                               }
-                                       }
-                               }
-                       }
+       if ((IO->HttpReq.httpcode >= 300) && (IO->HttpReq.httpcode < 400)  && (Ctx->RedirectUrl != NULL))
+       {
+               StrBuf *ErrMsg;
+               long lens[2];
+               const char *strs[2];
 
+               SetRSSState(IO, eRSSFailure);
+               ErrMsg = NewStrBuf();
+               if (IO) {
+                       syslog(LOG_INFO, "need a 200, got a %ld !", IO->HttpReq.httpcode);
                }
-               sock_close(sock);
+               strs[0] = ChrPtr(Ctx->Url);
+               lens[0] = StrLength(Ctx->Url);
+
+               strs[1] = ChrPtr(Ctx->rooms);
+               lens[1] = StrLength(Ctx->rooms);
+
+               if (IO->HttpReq.CurlError == NULL)
+                       IO->HttpReq.CurlError = "";
+
+               StrBufPrintf(ErrMsg,
+                    "Error while RSS-Aggregation Run of %s\n"
+                    " need a 200, got a %ld !\n"
+                    " Curl Error message: \n%s / %s\n"
+                    " Redirect header points to: %s\n"
+                    " Response text was: \n"
+                    " \n %s\n",
+                    ChrPtr(Ctx->Url),
+                    IO->HttpReq.httpcode,
+                    IO->HttpReq.errdesc,
+                    IO->HttpReq.CurlError,
+                    ChrPtr(Ctx->RedirectUrl),
+                    ChrPtr(IO->HttpReq.ReplyData)
+               );
+
+               CtdlAideFPMessage(
+                       ChrPtr(ErrMsg),
+                       "RSS Aggregation run failure",
+                       2, strs, (long*) &lens,
+                       CCID,
+                       IO->ID,
+                       EvGetNow(IO));
+               
+               FreeStrBuf(&ErrMsg);
+               syslog(LOG_DEBUG,
+                             "RSS feed returned an invalid http status code. <%s><HTTP %ld>",
+                             ChrPtr(Ctx->Url),
+                             IO->HttpReq.httpcode
+               );
+               return eAbort;
+       }
+       else if (IO->HttpReq.httpcode != 200)
+       {
+               StrBuf *ErrMsg;
+               long lens[2];
+               const char *strs[2];
+
+               SetRSSState(IO, eRSSFailure);
+               ErrMsg = NewStrBuf();
+               if (IO) {
+                       syslog(LOG_INFO, "need a 200, got a %ld !", IO->HttpReq.httpcode);
+               }
+               strs[0] = ChrPtr(Ctx->Url);
+               lens[0] = StrLength(Ctx->Url);
+
+               strs[1] = ChrPtr(Ctx->rooms);
+               lens[1] = StrLength(Ctx->rooms);
+
+               if (IO->HttpReq.CurlError == NULL)
+                       IO->HttpReq.CurlError = "";
+
+               StrBufPrintf(ErrMsg,
+                            "Error while RSS-Aggregation Run of %s\n"
+                            " need a 200, got a %ld !\n"
+                            " Curl Error message: \n%s / %s\n"
+                            " Response text was: \n"
+                            " \n %s\n",
+                            ChrPtr(Ctx->Url),
+                            IO->HttpReq.httpcode,
+                            IO->HttpReq.errdesc,
+                            IO->HttpReq.CurlError,
+                            ChrPtr(IO->HttpReq.ReplyData)
+                       );
+
+               CtdlAideFPMessage(
+                       ChrPtr(ErrMsg),
+                       "RSS Aggregation run failure",
+                       2, strs, (long*) &lens,
+                       CCID,
+                       IO->ID,
+                       EvGetNow(IO));
+               
+               FreeStrBuf(&ErrMsg);
+               syslog(LOG_DEBUG,
+                             "RSS feed returned an invalid http status code. <%s><HTTP %ld>",
+                             ChrPtr(Ctx->Url),
+                             IO->HttpReq.httpcode
+               );
+               return eAbort;
        }
-       else {
-               lprintf(CTDL_ERR, "Could not connect: %s\n", strerror(errno));
+
+       pCfg = &Ctx->Cfg;
+
+       while (pCfg != NULL)
+       {
+               UpdateLastKnownGood (pCfg, EvGetNow(IO));
+               if ((Ctx->roomlist_parts > 1) && 
+                   (it == NULL))
+               {
+                       it = GetNewHashPos(RSSFetchUrls, 0);
+               }
+               if (it != NULL)
+               {
+                       void *vptr;
+                       if (GetNextHashPos(Ctx->OtherQRnumbers, it, &len, &Key, &vptr))
+                               pCfg = vptr;
+                       else
+                               pCfg = NULL;
+               }
+               else 
+                       pCfg = NULL;
        }
+       DeleteHashPos (&it);
+
+       SetRSSState(IO, eRSSUT);
+
+       MD5Init(&md5context);
+
+       MD5Update(&md5context,
+                 (const unsigned char*)SKEY(IO->HttpReq.ReplyData));
+
+       MD5Update(&md5context,
+                 (const unsigned char*)SKEY(Ctx->Url));
+
+       MD5Final(rawdigest, &md5context);
+       guid = NewStrBufPlain(NULL,
+                             MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/);
+       StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN);
+       StrBufAppendBufPlain(guid, HKEY("_rssFM"), 0);
+       if (StrLength(guid) > 40)
+               StrBufCutAt(guid, 40, NULL);
+       /* Find out if we've already seen this item */
 
-       XML_ParserFree(xp);
-
-       /* Free the feed item data structure */
-       if (ri.guid != NULL) free(ri.guid);
-       ri.guid = NULL;
-       if (ri.title != NULL) free(ri.title);
-       ri.title = NULL;
-       if (ri.link != NULL) free(ri.link);
-       ri.link = NULL;
-       if (ri.description != NULL) free(ri.description);
-       ri.description = NULL;
-       if (ri.chardata_len > 0) {
-               free(ri.chardata);
-               ri.chardata = 0;
-               ri.chardata_len = 0;
+#ifndef DEBUG_RSS
+
+       if (CheckIfAlreadySeen("RSS Whole",
+                              guid,
+                              EvGetNow(IO),
+                              EvGetNow(IO) - USETABLE_ANTIEXPIRE,
+                              eUpdate,
+                              CCID, IO->ID)
+           != 0)
+       {
+               FreeStrBuf(&guid);
+
+               syslog(LOG_DEBUG, "RSS feed already seen. <%s>", ChrPtr(Ctx->Url));
+               return eAbort;
        }
+       FreeStrBuf(&guid);
+#endif
+       SetRSSState(IO, eRSSParsing);
+       return RSSAggregator_ParseReply(IO);
 }
 
+eNextState RSSAggregator_FinishHttp(AsyncIO *IO)
+{
+       return CurlQueueDBOperation(IO, RSSAggregator_AnalyseReply);
+}
+
+/*
+ * Begin a feed parse
+ */
+int rss_do_fetching(rss_aggregator *RSSAggr)
+{
+       AsyncIO         *IO = &RSSAggr->IO;
+       rss_item *ri;
+       time_t now;
+       CURLcode sta;
+       CURL *chnd;
+
+
+       now = time(NULL);
+
+       if ((RSSAggr->next_poll != 0) && (now < RSSAggr->next_poll))
+               return 0;
+
+       ri = (rss_item*) malloc(sizeof(rss_item));
+       memset(ri, 0, sizeof(rss_item));
+       RSSAggr->Item = ri;
+
+       if (! InitcURLIOStruct(&RSSAggr->IO,
+                              RSSAggr,
+                              "Citadel RSS Client",
+                              RSSAggregator_FinishHttp,
+                              RSSAggregator_Terminate,
+                              RSSAggregator_TerminateDB,
+                              RSSAggregator_ShutdownAbort))
+       {
+               syslog(LOG_INFO, "Unable to initialize libcurl.");
+               return 0;
+       }
+       chnd = IO->HttpReq.chnd;
+       OPT(HEADERDATA, IO);
+       OPT(HEADERFUNCTION, GetLocationString);
+       SetRSSState(IO, eRSSCreated);
+
+       safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host,
+                   ChrPtr(RSSAggr->Url),
+                   sizeof(((CitContext*)RSSAggr->IO.CitContext)->cs_host));
+
+       syslog(LOG_DEBUG, "Fetching RSS feed <%s>", ChrPtr(RSSAggr->Url));
+       ParseURL(&RSSAggr->IO.ConnectMe, RSSAggr->Url, 80);
+       CurlPrepareURL(RSSAggr->IO.ConnectMe);
+
+       SetRSSState(IO, eRSSFetching);
+       QueueCurlContext(&RSSAggr->IO);
+       return 1;
+}
 
 /*
  * Scan a room's netconfig to determine whether it is requesting any RSS feeds
  */
-void rssclient_scan_room(struct ctdlroom *qrbuf, void *data)
+void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
 {
-       char filename[PATH_MAX];
-       char buf[1024];
-       char instr[32];
-       FILE *fp;
-       char feedurl[256];
-       struct rssnetcfg *rncptr = NULL;
-       struct rssnetcfg *use_this_rncptr = NULL;
-       int len = 0;
-       char *ptr = NULL;
-
-       assoc_file_name(filename, sizeof filename, qrbuf, ctdl_netcfg_dir);
-
-       /* Only do net processing for rooms that have netconfigs */
-       fp = fopen(filename, "r");
-       if (fp == NULL) {
+       const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
+       rss_aggregator *RSSAggr = NULL;
+       rss_aggregator *use_this_RSSAggr = NULL;
+       void *vptr;
+
+       syslog(LOG_DEBUG, "rssclient_scan_room(%s)", qrbuf->QRname);
+       pthread_mutex_lock(&RSSQueueMutex);
+       if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
+       {
+               syslog(LOG_DEBUG,
+                             "rssclient: [%ld] %s already in progress.",
+                             qrbuf->QRnumber,
+                             qrbuf->QRname);
+               pthread_mutex_unlock(&RSSQueueMutex);
                return;
        }
+       pthread_mutex_unlock(&RSSQueueMutex);
+
+       if (server_shutting_down) return;
+
+       while (RSSCfg != NULL)
+       {
+               pthread_mutex_lock(&RSSQueueMutex);
+               GetHash(RSSFetchUrls,
+                       SKEY(RSSCfg->Url),
+                       &vptr);
+
+               use_this_RSSAggr = (rss_aggregator *)vptr;
+               if (use_this_RSSAggr != NULL)
+               {
+                       pRSSConfig *pRSSCfg;
+
+                       StrBufAppendBufPlain(
+                               use_this_RSSAggr->rooms,
+                               qrbuf->QRname,
+                               -1, 0);
+                       if (use_this_RSSAggr->roomlist_parts==1)
+                       {
+                               use_this_RSSAggr->OtherQRnumbers
+                                       = NewHash(1, lFlathash);
+                       }
 
-       while (fgets(buf, sizeof buf, fp) != NULL) {
-               buf[strlen(buf)-1] = 0;
+                       pRSSCfg = (pRSSConfig *) malloc(sizeof(pRSSConfig));
 
-               extract_token(instr, buf, 0, '|', sizeof instr);
-               if (!strcasecmp(instr, "rssclient")) {
-                       extract_token(feedurl, buf, 1, '|', sizeof feedurl);
+                       pRSSCfg->QRnumber = qrbuf->QRnumber;
+                       pRSSCfg->pCfg = RSSCfg;
 
-                       /* If any other rooms have requested the same feed, then we will just add this
-                        * room to the target list for that client request.
-                        */
-                       for (rncptr=rnclist; rncptr!=NULL; rncptr=rncptr->next) {
-                               if (!strcmp(rncptr->url, feedurl)) {
-                                       use_this_rncptr = rncptr;
-                               }
-                       }
+                       Put(use_this_RSSAggr->OtherQRnumbers,
+                           LKEY(qrbuf->QRnumber),
+                           pRSSCfg,
+                           NULL);
+                       use_this_RSSAggr->roomlist_parts++;
 
-                       /* Otherwise create a new client request */
-                       if (use_this_rncptr == NULL) {
-                               rncptr = (struct rssnetcfg *) malloc(sizeof(struct rssnetcfg));
-                               if (rncptr != NULL) {
-                                       rncptr->next = rnclist;
-                                       safestrncpy(rncptr->url, feedurl, sizeof rncptr->url);
-                                       rncptr->rooms = NULL;
-                                       rnclist = rncptr;
-                                       use_this_rncptr = rncptr;
-                               }
-                       }
+                       pthread_mutex_unlock(&RSSQueueMutex);
 
-                       /* 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;
-                                       }
-                               }
-                       }
+                       RSSCfg = RSSCfg->next;
+                       continue;
                }
+               pthread_mutex_unlock(&RSSQueueMutex);
 
-       }
+               RSSAggr = (rss_aggregator *) malloc(
+                       sizeof(rss_aggregator));
+
+               memset (RSSAggr, 0, sizeof(rss_aggregator));
+               RSSAggr->Cfg.QRnumber = qrbuf->QRnumber;
+               RSSAggr->Cfg.pCfg = RSSCfg;
+               RSSAggr->roomlist_parts = 1;
+               RSSAggr->Url = NewStrBufDup(RSSCfg->Url);
+
+               RSSAggr->ItemType = RSS_UNSET;
 
-       fclose(fp);
+               RSSAggr->rooms = NewStrBufPlain(
+                       qrbuf->QRname, -1);
 
+               pthread_mutex_lock(&RSSQueueMutex);
+
+               Put(RSSFetchUrls,
+                   SKEY(RSSAggr->Url),
+                   RSSAggr,
+                   DeleteRssCfg);
+
+               pthread_mutex_unlock(&RSSQueueMutex);
+               RSSCfg = RSSCfg->next;
+       }
 }
 
 /*
  * Scan for rooms that have RSS client requests configured
  */
 void rssclient_scan(void) {
-       static time_t last_run = 0L;
-       static int doing_rssclient = 0;
-       struct rssnetcfg *rptr = NULL;
+       int RSSRoomCount, RSSCount;
+       rss_aggregator *rptr = NULL;
+       void *vrptr = NULL;
+       HashPos *it;
+       long len;
+       const char *Key;
+       time_t now = time(NULL);
+
+       /* Run no more than once every 15 minutes. */
+       if ((now - last_run) < 900) {
+               syslog(LOG_DEBUG,
+                             "Client: polling interval not yet reached; last run was %ldm%lds ago",
+                             ((now - last_run) / 60),
+                             ((now - last_run) % 60)
+               );
+               return;
+       }
 
        /*
-        * Run RSS client no more frequently than once every n seconds
+        * This is a simple concurrency check to make sure only one rssclient
+        * run is done at a time.
         */
-       if ( (time(NULL) - last_run) < config.c_net_freq ) {
+       pthread_mutex_lock(&RSSQueueMutex);
+       RSSCount = GetCount(RSSFetchUrls);
+       RSSRoomCount = GetCount(RSSQueueRooms);
+       pthread_mutex_unlock(&RSSQueueMutex);
+
+       if ((RSSRoomCount > 0) || (RSSCount > 0)) {
+               syslog(LOG_DEBUG,
+                             "rssclient: concurrency check failed; %d rooms and %d url's are queued",
+                             RSSRoomCount, RSSCount
+                       );
                return;
        }
 
-       /*
-        * This is a simple concurrency check to make sure only one rssclient run
-        * is done at a time.  We could do this with a mutex, but since we
-        * don't really require extremely fine granularity here, we'll do it
-        * with a static variable instead.
-        */
-       if (doing_rssclient) return;
-       doing_rssclient = 1;
-
-       lprintf(CTDL_DEBUG, "rssclient started\n");
-       ForEachRoom(rssclient_scan_room, NULL);
-
-       while (rnclist != NULL) {
-               rss_do_fetching(rnclist->url, rnclist->rooms);
-               rptr = rnclist;
-               rnclist = rnclist->next;
-               if (rptr->rooms != NULL) free(rptr->rooms);
-               free(rptr);
+       become_session(&rss_CC);
+       syslog(LOG_DEBUG, "rssclient started");
+       CtdlForEachNetCfgRoom(rssclient_scan_room, NULL, rssclient);
+
+       if (GetCount(RSSFetchUrls) > 0)
+       {
+               pthread_mutex_lock(&RSSQueueMutex);
+               syslog(LOG_DEBUG,
+                              "rssclient starting %d Clients",
+                              GetCount(RSSFetchUrls));
+               
+               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);
+       }
+       else {
+               syslog(LOG_DEBUG, "Nothing to do.");
        }
 
-       lprintf(CTDL_DEBUG, "rssclient ended\n");
-       last_run = time(NULL);
-       doing_rssclient = 0;
+       syslog(LOG_DEBUG, "rssclient ended");
+       return;
+}
+
+void rss_cleanup(void)
+{
+       /* citthread_mutex_destroy(&RSSQueueMutex); TODO */
+       DeleteHash(&RSSFetchUrls);
+       DeleteHash(&RSSQueueRooms);
 }
 
+void LogDebugEnableRSSClient(const int n)
+{
+       RSSClientDebugEnabled = n;
+}
+
+
+typedef struct __RSSVetoInfo {
+       StrBuf *ErrMsg;
+       time_t Now;
+       int Veto;
+}RSSVetoInfo;
+
+void rssclient_veto_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
+{
+       RSSVetoInfo *Info = (RSSVetoInfo *) data;
+       const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
+
+       while (RSSCfg != NULL)
+       {
+               if ((RSSCfg->last_known_good != 0) &&
+                   (RSSCfg->last_known_good + USETABLE_ANTIEXPIRE < Info->Now))
+               {
+                       StrBufAppendPrintf(Info->ErrMsg,
+                                          "RSS feed not seen for a %d days:: <",
+                                          (Info->Now - RSSCfg->last_known_good) / (24 * 60 * 60));
+
+                       StrBufAppendBuf(Info->ErrMsg, RSSCfg->Url, 0);
+                       StrBufAppendBufPlain(Info->ErrMsg, HKEY(">\n"), 0);
+               }
+               RSSCfg = RSSCfg->next;
+       }
+}
+
+int RSSCheckUsetableVeto(StrBuf *ErrMsg)
+{
+       RSSVetoInfo Info;
+
+       Info.ErrMsg = ErrMsg;
+       Info.Now = time (NULL);
+       Info.Veto = 0;
+
+       CtdlForEachNetCfgRoom(rssclient_veto_scan_room, &Info, rssclient);
+
+       return Info.Veto;;
+}
+
+
+
+
+void ParseRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *Line, const char *LinePos, OneRoomNetCfg *OneRNCFG)
+{
+       RSSCfgLine *RSSCfg;
+
+       RSSCfg = (RSSCfgLine *) malloc (sizeof(RSSCfgLine));
+       RSSCfg->Url = NewStrBufPlain (NULL, StrLength (Line));
+       
+
+       StrBufExtract_NextToken(RSSCfg->Url, Line, &LinePos, '|');
+       RSSCfg->last_known_good = StrBufExtractNext_long(Line, &LinePos, '|');
+
+
+       RSSCfg->next = (RSSCfgLine *)OneRNCFG->NetConfigs[ThisOne->C];
+       OneRNCFG->NetConfigs[ThisOne->C] = (RoomNetCfgLine*) RSSCfg;
+}
+
+void SerializeRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *OutputBuffer, OneRoomNetCfg *RNCfg, RoomNetCfgLine *data)
+{
+       RSSCfgLine *RSSCfg = (RSSCfgLine*) data;
+
+       StrBufAppendBufPlain(OutputBuffer, CKEY(ThisOne->Str), 0);
+       StrBufAppendBufPlain(OutputBuffer, HKEY("|"), 0);
+       StrBufAppendBufPlain(OutputBuffer, SKEY(RSSCfg->Url), 0);
+       StrBufAppendPrintf(OutputBuffer, "|%ld\n", RSSCfg->last_known_good);
+}
+
+void DeleteRSSClientCfgLine(const CfgLineType *ThisOne, RoomNetCfgLine **data)
+{
+       RSSCfgLine *RSSCfg = (RSSCfgLine*) *data;
+
+       FreeStrBuf(&RSSCfg->Url);
+       free(*data);
+       *data = NULL;
+}
 
-#endif /* HAVE_EXPAT */
 
 CTDL_MODULE_INIT(rssclient)
 {
-#ifdef HAVE_EXPAT
-       CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER);
-#else
-        lprintf(CTDL_INFO, "This server is missing the Expat XML parser.  RSS client will be disabled.\n");
-#endif
-       /* return our Subversion id for the Log */
-        return "$Id: serv_rssclient.c 5652 2007-10-29 20:14:48Z ajc $";
+       if (!threading)
+       {
+               CtdlRegisterTDAPVetoHook (RSSCheckUsetableVeto, CDB_USETABLE, 0);
+
+               CtdlREGISTERRoomCfgType(rssclient, ParseRSSClientCfgLine, 0, 1, SerializeRSSClientCfgLine, DeleteRSSClientCfgLine);
+               pthread_mutex_init(&RSSQueueMutex, NULL);
+               RSSQueueRooms = NewHash(1, lFlathash);
+               RSSFetchUrls = NewHash(1, NULL);
+               syslog(LOG_INFO, "%s\n", curl_version());
+               CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER, PRIO_AGGR + 300);
+               CtdlRegisterEVCleanupHook(rss_cleanup);
+               CtdlRegisterDebugFlagHook(HKEY("rssclient"), LogDebugEnableRSSClient, &RSSClientDebugEnabled);
+       }
+       else
+       {
+               CtdlFillSystemContext(&rss_CC, "rssclient");
+       }
+       return "rssclient";
 }