SEEN-Database: refactor database interface for remembering whether we already aggrega...

[citadel.git] / citadel / modules / rssclient / serv_rssclient.c

diff --git a/citadel/modules/rssclient/serv_rssclient.c b/citadel/modules/rssclient/serv_rssclient.c
index c2d95d34a88de0e68b0510524129fcfccafa16ba..7cc8160fbb09fe172e8f4bbec9224c4560bc1c7f 100644 (file)
--- a/citadel/modules/rssclient/serv_rssclient.c
+++ b/citadel/modules/rssclient/serv_rssclient.c
@@ -164,7 +164,9 @@ void DeleteRssCfg(void *vptr)
 {
        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);
@@ -369,10 +371,8 @@ eNextState RSSSaveMessage(AsyncIO *IO)
        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,
@@ -387,27 +387,23 @@ eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
 {
        const char *Key;
        long len;
-       struct cdbdata *cdbut;
        rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
 
        /* Find out if we've already seen this item */
-       strcpy(Ctx->ThisMsg->ut.ut_msgid,
-              ChrPtr(Ctx->ThisMsg->MsgGUID)); /// TODO
-       Ctx->ThisMsg->ut.ut_timestamp = time(NULL);
-
-       cdbut = cdb_fetch(CDB_USETABLE, SKEY(Ctx->ThisMsg->MsgGUID));
+// todo: expiry?
 #ifndef DEBUG_RSS
-       if (cdbut != NULL) {
+       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) );
 
                if (GetNextHashPos(Ctx->Messages,
                                   Ctx->Pos,
@@ -425,15 +421,14 @@ eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
                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)
@@ -462,7 +457,9 @@ eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
                CtdlAideFPMessage(
                        ChrPtr(ErrMsg),
                        "RSS Aggregation run failure",
-                       2, strs, (long*) &lens);
+                       2, strs, (long*) &lens,
+                       IO->Now,
+                       IO->ID, CCID);
                FreeStrBuf(&ErrMsg);
                return eAbort;
        }
@@ -483,25 +480,22 @@ eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
        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);
+
+               return eAbort;
+       }
        FreeStrBuf(&guid);
-       if (cdbut != NULL) return eAbort;
 #endif
        return RSSAggregator_ParseReply(IO);
 }
@@ -559,7 +553,6 @@ int rss_do_fetching(rss_aggregator *RSSAggr)
 void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
 {
        const RoomNetCfgLine *pLine;
-       rss_room_counter *Count = NULL;
        rss_aggregator *RSSAggr = NULL;
        rss_aggregator *use_this_RSSAggr = NULL;
        void *vptr;
@@ -578,24 +571,23 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneR
 
        if (server_shutting_down) return;
 
-       pLine = OneRNCFG->NetConfigs[pop3client];
+       pLine = OneRNCFG->NetConfigs[rssclient];
 
        while (pLine != NULL)
        {
-               if (Count == NULL)
-               {
-                       Count = malloc(
-                               sizeof(rss_room_counter));
-                       Count->count = 0;
-               }
-               Count->count ++;
+               const char *lPtr = NULL;
+
                RSSAggr = (rss_aggregator *) malloc(
                        sizeof(rss_aggregator));
 
                memset (RSSAggr, 0, sizeof(rss_aggregator));
                RSSAggr->QRnumber = qrbuf->QRnumber;
                RSSAggr->roomlist_parts = 1;
-               RSSAggr->Url = NewStrBufDup(pLine->Value[1]);
+               RSSAggr->Url = NewStrBufPlain(NULL, StrLength(pLine->Value[0]));
+               StrBufExtract_NextToken(RSSAggr->Url,
+                                       pLine->Value[0],
+                                       &lPtr,
+                                       '|');
 
                pthread_mutex_lock(&RSSQueueMutex);
                GetHash(RSSFetchUrls,
@@ -628,6 +620,7 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneR
                        FreeStrBuf(&RSSAggr->Url);
                        free(RSSAggr);
                        RSSAggr = NULL;
+                       pLine = pLine->next;
                        continue;
                }
                pthread_mutex_unlock(&RSSQueueMutex);
@@ -645,6 +638,7 @@ void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneR
                    DeleteRssCfg);
 
                pthread_mutex_unlock(&RSSQueueMutex);
+               pLine = pLine->next;
        }
 }