DBGLOG(LEVEL) syslog(LEVEL, "IO[%ld][%ld]RSS" FORMAT, \
IO->ID, N)
+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")}
+};
+
+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);
+}
+
void DeleteRoomReference(long QRnumber)
{
HashPos *At;
{
rss_aggregator *RSSAggr = (rss_aggregator *)vptr;
AsyncIO *IO = &RSSAggr->IO;
- EVRSSCM_syslog(LOG_DEBUG, "RSS: destroying\n");
+
+ if (IO->CitContext != NULL)
+ EVRSSCM_syslog(LOG_DEBUG, "RSS: destroying\n");
FreeStrBuf(&RSSAggr->Url);
FreeStrBuf(&RSSAggr->rooms);
CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0);
/* write the uidl to the use table so we don't store this item again */
- cdb_store(CDB_USETABLE,
- SKEY(RSSAggr->ThisMsg->MsgGUID),
- &RSSAggr->ThisMsg->ut,
- sizeof(struct UseTable) );
+
+ CheckIfAlreadySeen("RSS Item Insert", RSSAggr->ThisMsg->MsgGUID, IO->Now, 0, eWrite, IO->ID, CCID);
if (GetNextHashPos(RSSAggr->Messages,
RSSAggr->Pos,
{
const char *Key;
long len;
- struct cdbdata *cdbut;
rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
/* Find out if we've already seen this item */
- strcpy(Ctx->ThisMsg->ut.ut_msgid,
- ChrPtr(Ctx->ThisMsg->MsgGUID)); /// TODO
- Ctx->ThisMsg->ut.ut_timestamp = time(NULL);
-
- cdbut = cdb_fetch(CDB_USETABLE, SKEY(Ctx->ThisMsg->MsgGUID));
+// todo: expiry?
#ifndef DEBUG_RSS
- if (cdbut != NULL) {
+ SetRSSState(IO, eRSSUT);
+ if (CheckIfAlreadySeen("RSS Item Seen",
+ Ctx->ThisMsg->MsgGUID,
+ IO->Now,
+ IO->Now - USETABLE_ANTIEXPIRE,
+ eCheckUpdate,
+ IO->ID, CCID)
+ != 0)
+ {
/* Item has already been seen */
EVRSSC_syslog(LOG_DEBUG,
"%s has already been seen\n",
ChrPtr(Ctx->ThisMsg->MsgGUID));
- cdb_free(cdbut);
-
- /* rewrite the record anyway, to update the timestamp */
- cdb_store(CDB_USETABLE,
- SKEY(Ctx->ThisMsg->MsgGUID),
- &Ctx->ThisMsg->ut, sizeof(struct UseTable) );
+ SetRSSState(IO, eRSSParsing);
if (GetNextHashPos(Ctx->Messages,
Ctx->Pos,
else
#endif
{
+ SetRSSState(IO, eRSSParsing);
+
NextDBOperation(IO, RSSSaveMessage);
return eSendMore;
}
+ return eSendMore;
}
eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
{
- struct UseTable ut;
u_char rawdigest[MD5_DIGEST_LEN];
struct MD5Context md5context;
StrBuf *guid;
- struct cdbdata *cdbut;
rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
if (IO->HttpReq.httpcode != 200)
long lens[2];
const char *strs[2];
+ SetRSSState(IO, eRSSFailure);
ErrMsg = NewStrBuf();
EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n",
IO->HttpReq.httpcode);
CtdlAideFPMessage(
ChrPtr(ErrMsg),
"RSS Aggregation run failure",
- 2, strs, (long*) &lens);
+ 2, strs, (long*) &lens,
+ IO->Now,
+ IO->ID, CCID);
+
FreeStrBuf(&ErrMsg);
+ EVRSSC_syslog(LOG_DEBUG,
+ "RSS feed returned an invalid http status code. <%s><HTTP %ld>\n",
+ ChrPtr(Ctx->Url),
+ IO->HttpReq.httpcode);
return eAbort;
}
+ SetRSSState(IO, eRSSUT);
MD5Init(&md5context);
if (StrLength(guid) > 40)
StrBufCutAt(guid, 40, NULL);
/* Find out if we've already seen this item */
- memcpy(ut.ut_msgid, SKEY(guid));
- ut.ut_timestamp = time(NULL);
- cdbut = cdb_fetch(CDB_USETABLE, SKEY(guid));
#ifndef DEBUG_RSS
- if (cdbut != NULL) {
- /* Item has already been seen */
- EVRSSC_syslog(LOG_DEBUG,
- "%s has already been seen\n",
- ChrPtr(Ctx->Url));
- cdb_free(cdbut);
- }
- /* rewrite the record anyway, to update the timestamp */
- cdb_store(CDB_USETABLE,
- SKEY(guid),
- &ut, sizeof(struct UseTable) );
+ if (CheckIfAlreadySeen("RSS Whole",
+ guid,
+ IO->Now,
+ IO->Now - USETABLE_ANTIEXPIRE,
+ eCheckUpdate,
+ IO->ID, CCID)
+ != 0)
+ {
+ FreeStrBuf(&guid);
+
+ EVRSSC_syslog(LOG_DEBUG, "RSS feed already seen. <%s>\n", ChrPtr(Ctx->Url));
+ return eAbort;
+ }
FreeStrBuf(&guid);
- if (cdbut != NULL) return eAbort;
#endif
+ SetRSSState(IO, eRSSParsing);
return RSSAggregator_ParseReply(IO);
}
EVRSSCM_syslog(LOG_ALERT, "Unable to initialize libcurl.\n");
return 0;
}
+ SetRSSState(IO, eRSSCreated);
safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host,
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)
{
- StrBuf *CfgData=NULL;
- StrBuf *CfgType;
- StrBuf *Line;
- rss_room_counter *Count = NULL;
- struct stat statbuf;
- char filename[PATH_MAX];
- int fd;
- int Done;
+ const RoomNetCfgLine *pLine;
rss_aggregator *RSSAggr = NULL;
rss_aggregator *use_this_RSSAggr = NULL;
void *vptr;
- const char *CfgPtr, *lPtr;
- const char *Err;
pthread_mutex_lock(&RSSQueueMutex);
if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
}
pthread_mutex_unlock(&RSSQueueMutex);
- assoc_file_name(filename, sizeof filename, qrbuf, ctdl_netcfg_dir);
+ if (server_shutting_down) return;
- if (server_shutting_down)
- return;
+ pLine = OneRNCFG->NetConfigs[rssclient];
- /* Only do net processing for rooms that have netconfigs */
- fd = open(filename, 0);
- if (fd <= 0) {
- /* syslog(LOG_DEBUG,
- "rssclient: %s no config.\n",
- qrbuf->QRname); */
- return;
- }
-
- if (server_shutting_down)
- return;
-
- if (fstat(fd, &statbuf) == -1) {
- EVRSSQ_syslog(LOG_DEBUG,
- "ERROR: could not stat configfile '%s' - %s\n",
- filename,
- strerror(errno));
- return;
- }
+ while (pLine != NULL)
+ {
+ const char *lPtr = NULL;
- if (server_shutting_down)
- return;
+ RSSAggr = (rss_aggregator *) malloc(
+ sizeof(rss_aggregator));
- CfgData = NewStrBufPlain(NULL, statbuf.st_size + 1);
+ memset (RSSAggr, 0, sizeof(rss_aggregator));
+ RSSAggr->QRnumber = qrbuf->QRnumber;
+ RSSAggr->roomlist_parts = 1;
+ RSSAggr->Url = NewStrBufPlain(NULL, StrLength(pLine->Value[0]));
+ StrBufExtract_NextToken(RSSAggr->Url,
+ pLine->Value[0],
+ &lPtr,
+ '|');
- if (StrBufReadBLOB(CfgData, &fd, 1, statbuf.st_size, &Err) < 0) {
- close(fd);
- FreeStrBuf(&CfgData);
- EVRSSQ_syslog(LOG_ERR, "ERROR: reading config '%s' - %s<br>\n",
- filename, strerror(errno));
- return;
- }
- close(fd);
- if (server_shutting_down)
- return;
+ pthread_mutex_lock(&RSSQueueMutex);
+ GetHash(RSSFetchUrls,
+ SKEY(RSSAggr->Url),
+ &vptr);
- CfgPtr = NULL;
- CfgType = NewStrBuf();
- Line = NewStrBufPlain(NULL, StrLength(CfgData));
- Done = 0;
- while (!Done)
- {
- Done = StrBufSipLine(Line, CfgData, &CfgPtr) == 0;
- if (StrLength(Line) > 0)
+ use_this_RSSAggr = (rss_aggregator *)vptr;
+ if (use_this_RSSAggr != NULL)
{
- lPtr = NULL;
- StrBufExtract_NextToken(CfgType, Line, &lPtr, '|');
- if (!strcasecmp("rssclient", ChrPtr(CfgType)))
+ long *QRnumber;
+ StrBufAppendBufPlain(
+ use_this_RSSAggr->rooms,
+ qrbuf->QRname,
+ -1, 0);
+ if (use_this_RSSAggr->roomlist_parts==1)
{
- if (Count == NULL)
- {
- Count = malloc(
- sizeof(rss_room_counter));
- Count->count = 0;
- }
- Count->count ++;
- RSSAggr = (rss_aggregator *) malloc(
- sizeof(rss_aggregator));
-
- memset (RSSAggr, 0, sizeof(rss_aggregator));
- RSSAggr->QRnumber = qrbuf->QRnumber;
- RSSAggr->roomlist_parts = 1;
- RSSAggr->Url = NewStrBuf();
-
- StrBufExtract_NextToken(RSSAggr->Url,
- Line,
- &lPtr,
- '|');
-
- pthread_mutex_lock(&RSSQueueMutex);
- GetHash(RSSFetchUrls,
- SKEY(RSSAggr->Url),
- &vptr);
-
- use_this_RSSAggr = (rss_aggregator *)vptr;
- if (use_this_RSSAggr != NULL)
- {
- long *QRnumber;
- StrBufAppendBufPlain(
- use_this_RSSAggr->rooms,
- qrbuf->QRname,
- -1, 0);
- if (use_this_RSSAggr->roomlist_parts==1)
- {
- use_this_RSSAggr->OtherQRnumbers
- = NewHash(1, lFlathash);
- }
- QRnumber = (long*)malloc(sizeof(long));
- *QRnumber = qrbuf->QRnumber;
- Put(use_this_RSSAggr->OtherQRnumbers,
- LKEY(qrbuf->QRnumber),
- QRnumber,
- NULL);
- use_this_RSSAggr->roomlist_parts++;
-
- pthread_mutex_unlock(&RSSQueueMutex);
-
- FreeStrBuf(&RSSAggr->Url);
- free(RSSAggr);
- RSSAggr = NULL;
- continue;
- }
- pthread_mutex_unlock(&RSSQueueMutex);
-
- RSSAggr->ItemType = RSS_UNSET;
-
- RSSAggr->rooms = NewStrBufPlain(
- qrbuf->QRname, -1);
-
- pthread_mutex_lock(&RSSQueueMutex);
-
- Put(RSSFetchUrls,
- SKEY(RSSAggr->Url),
- RSSAggr,
- DeleteRssCfg);
-
- pthread_mutex_unlock(&RSSQueueMutex);
+ use_this_RSSAggr->OtherQRnumbers
+ = NewHash(1, lFlathash);
}
+ QRnumber = (long*)malloc(sizeof(long));
+ *QRnumber = qrbuf->QRnumber;
+ Put(use_this_RSSAggr->OtherQRnumbers,
+ LKEY(qrbuf->QRnumber),
+ QRnumber,
+ NULL);
+ use_this_RSSAggr->roomlist_parts++;
+
+ pthread_mutex_unlock(&RSSQueueMutex);
+
+ FreeStrBuf(&RSSAggr->Url);
+ free(RSSAggr);
+ RSSAggr = NULL;
+ pLine = pLine->next;
+ continue;
}
- }
- if (Count != NULL)
- {
- Count->QRnumber = qrbuf->QRnumber;
+ pthread_mutex_unlock(&RSSQueueMutex);
+
+ RSSAggr->ItemType = RSS_UNSET;
+
+ RSSAggr->rooms = NewStrBufPlain(
+ qrbuf->QRname, -1);
+
pthread_mutex_lock(&RSSQueueMutex);
- EVRSSQ_syslog(LOG_DEBUG, "client: [%ld] %s now starting.\n",
- qrbuf->QRnumber, qrbuf->QRname);
- Put(RSSQueueRooms, LKEY(qrbuf->QRnumber), Count, NULL);
+
+ Put(RSSFetchUrls,
+ SKEY(RSSAggr->Url),
+ RSSAggr,
+ DeleteRssCfg);
+
pthread_mutex_unlock(&RSSQueueMutex);
+ pLine = pLine->next;
}
- FreeStrBuf(&CfgData);
- FreeStrBuf(&CfgType);
- FreeStrBuf(&Line);
}
/*
}
become_session(&rss_CC);
- EVRSSQM_syslog(LOG_DEBUG, "rssclient started\n");
- CtdlForEachRoom(rssclient_scan_room, NULL);
+ EVRSSQM_syslog(LOG_DEBUG, "rssclient started");
+ CtdlForEachNetCfgRoom(rssclient_scan_room, NULL, rssclient);
- pthread_mutex_lock(&RSSQueueMutex);
-
- it = GetNewHashPos(RSSFetchUrls, 0);
- while (!server_shutting_down &&
- GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) &&
- (vrptr != NULL)) {
- rptr = (rss_aggregator *)vrptr;
- if (!rss_do_fetching(rptr))
- UnlinkRSSAggregator(rptr);
+ if (GetCount(RSSFetchUrls) > 0)
+ {
+ pthread_mutex_lock(&RSSQueueMutex);
+ EVRSSQ_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);
}
- DeleteHashPos(&it);
- pthread_mutex_unlock(&RSSQueueMutex);
+ else
+ EVRSSQM_syslog(LOG_DEBUG, "Nothing to do.");
EVRSSQM_syslog(LOG_DEBUG, "rssclient ended\n");
return;