TDAP: align rss config structs with the primary type.
[citadel] / citadel / modules / rssclient / serv_rssclient.c
1 /*
2  * Bring external RSS feeds into rooms.
3  *
4  * Copyright (c) 2007-2012 by the citadel.org team
5  *
6  * This program is open source software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 3.
8  *
9  * This program is distributed in the hope that it will be useful,
10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.See the
12  * GNU General Public License for more details.
13  */
14
15 #include <stdlib.h>
16 #include <unistd.h>
17 #include <stdio.h>
18
19 #if TIME_WITH_SYS_TIME
20 # include <sys/time.h>
21 # include <time.h>
22 #else
23 # if HAVE_SYS_TIME_H
24 #include <sys/time.h>
25 # else
26 #include <time.h>
27 # endif
28 #endif
29
30 #include <ctype.h>
31 #include <string.h>
32 #include <errno.h>
33 #include <sys/types.h>
34 #include <sys/stat.h>
35 #include <expat.h>
36 #include <curl/curl.h>
37 #include <libcitadel.h>
38 #include "citadel.h"
39 #include "server.h"
40 #include "citserver.h"
41 #include "support.h"
42 #include "config.h"
43 #include "threads.h"
44 #include "ctdl_module.h"
45 #include "msgbase.h"
46 #include "parsedate.h"
47 #include "database.h"
48 #include "citadel_dirs.h"
49 #include "md5.h"
50 #include "context.h"
51 #include "event_client.h"
52 #include "rss_atom_parser.h"
53
54
55 #define TMP_MSGDATA 0xFF
56 #define TMP_SHORTER_URL_OFFSET 0xFE
57 #define TMP_SHORTER_URLS 0xFD
58
59 time_t last_run = 0L;
60
61 pthread_mutex_t RSSQueueMutex; /* locks the access to the following vars: */
62 HashList *RSSQueueRooms = NULL; /* rss_room_counter */
63 HashList *RSSFetchUrls = NULL; /*->rss_aggregator;->RefCount access locked*/
64
65 eNextState RSSAggregator_Terminate(AsyncIO *IO);
66 eNextState RSSAggregator_TerminateDB(AsyncIO *IO);
67 eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO);
68 struct CitContext rss_CC;
69
70 struct rssnetcfg *rnclist = NULL;
71 int RSSClientDebugEnabled = 0;
72 #define N ((rss_aggregator*)IO->Data)->Cfg.QRnumber
73
74 #define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (RSSClientDebugEnabled != 0))
75
76 #define EVRSSC_syslog(LEVEL, FORMAT, ...)                               \
77         DBGLOG(LEVEL) syslog(LEVEL,                                     \
78                              "IO[%ld]CC[%d][%ld]RSS" FORMAT,            \
79                              IO->ID, CCID, N, __VA_ARGS__)
80
81 #define EVRSSCM_syslog(LEVEL, FORMAT)                                   \
82         DBGLOG(LEVEL) syslog(LEVEL,                                     \
83                              "IO[%ld]CC[%d][%ld]RSS" FORMAT,            \
84                              IO->ID, CCID, N)
85
86 #define EVRSSQ_syslog(LEVEL, FORMAT, ...)                               \
87         DBGLOG(LEVEL) syslog(LEVEL, "RSS" FORMAT,                       \
88                              __VA_ARGS__)
89 #define EVRSSQM_syslog(LEVEL, FORMAT)                   \
90         DBGLOG(LEVEL) syslog(LEVEL, "RSS" FORMAT)
91
92 #define EVRSSCSM_syslog(LEVEL, FORMAT)                                  \
93         DBGLOG(LEVEL) syslog(LEVEL, "IO[%ld][%ld]RSS" FORMAT,           \
94                              IO->ID, N)
95
96 typedef enum _RSSState {
97         eRSSCreated,
98         eRSSFetching,
99         eRSSFailure,
100         eRSSParsing,
101         eRSSUT
102 } RSSState;
103 ConstStr RSSStates[] = {
104         {HKEY("Aggregator created")},
105         {HKEY("Fetching content")},
106         {HKEY("Failed")},
107         {HKEY("parsing content")},
108         {HKEY("checking usetable")}
109 };
110
111 static void SetRSSState(AsyncIO *IO, RSSState State)
112 {
113         CitContext* CCC = IO->CitContext;
114         if (CCC != NULL)
115                 memcpy(CCC->cs_clientname, RSSStates[State].Key, RSSStates[State].len + 1);
116 }
117
118 void DeleteRoomReference(long QRnumber)
119 {
120         HashPos *At;
121         long HKLen;
122         const char *HK;
123         void *vData = NULL;
124         rss_room_counter *pRoomC;
125
126         At = GetNewHashPos(RSSQueueRooms, 0);
127
128         if (GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At))
129         {
130                 GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData);
131                 if (vData != NULL)
132                 {
133                         pRoomC = (rss_room_counter *) vData;
134                         pRoomC->count --;
135                         if (pRoomC->count == 0)
136                                 DeleteEntryFromHash(RSSQueueRooms, At);
137                 }
138         }
139         DeleteHashPos(&At);
140 }
141
142 void UnlinkRooms(rss_aggregator *RSSAggr)
143 {
144         DeleteRoomReference(RSSAggr->Cfg.QRnumber);
145         if (RSSAggr->OtherQRnumbers != NULL)
146         {
147                 long HKLen;
148                 const char *HK;
149                 HashPos *At;
150                 void *vData;
151
152                 At = GetNewHashPos(RSSAggr->OtherQRnumbers, 0);
153                 while (! server_shutting_down &&
154                        GetNextHashPos(RSSAggr->OtherQRnumbers,
155                                       At,
156                                       &HKLen, &HK,
157                                       &vData) &&
158                        (vData != NULL))
159                 {
160                         pRSSConfig *Data = (pRSSConfig*) vData;
161                         DeleteRoomReference(Data->QRnumber);
162                 }
163
164                 DeleteHashPos(&At);
165         }
166 }
167
168 void UnlinkRSSAggregator(rss_aggregator *RSSAggr)
169 {
170         HashPos *At;
171
172         pthread_mutex_lock(&RSSQueueMutex);
173         UnlinkRooms(RSSAggr);
174
175         At = GetNewHashPos(RSSFetchUrls, 0);
176         if (GetHashPosFromKey(RSSFetchUrls, SKEY(RSSAggr->Url), At))
177         {
178                 DeleteEntryFromHash(RSSFetchUrls, At);
179         }
180         DeleteHashPos(&At);
181         last_run = time(NULL);
182         pthread_mutex_unlock(&RSSQueueMutex);
183 }
184
185 void DeleteRssCfg(void *vptr)
186 {
187         rss_aggregator *RSSAggr = (rss_aggregator *)vptr;
188         AsyncIO *IO = &RSSAggr->IO;
189
190         if (IO->CitContext != NULL)
191                 EVRSSCM_syslog(LOG_DEBUG, "RSS: destroying\n");
192
193         FreeStrBuf(&RSSAggr->Url);
194         FreeStrBuf(&RSSAggr->rooms);
195         FreeStrBuf(&RSSAggr->CData);
196         FreeStrBuf(&RSSAggr->Key);
197         DeleteHash(&RSSAggr->OtherQRnumbers);
198
199         DeleteHashPos (&RSSAggr->Pos);
200         DeleteHash (&RSSAggr->Messages);
201         if (RSSAggr->recp.recp_room != NULL)
202                 free(RSSAggr->recp.recp_room);
203
204
205         if (RSSAggr->Item != NULL)
206         {
207                 flush_rss_item(RSSAggr->Item);
208
209                 free(RSSAggr->Item);
210         }
211
212         FreeAsyncIOContents(&RSSAggr->IO);
213         memset(RSSAggr, 0, sizeof(rss_aggregator));
214         free(RSSAggr);
215 }
216
217 eNextState RSSAggregator_Terminate(AsyncIO *IO)
218 {
219         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
220
221         EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
222
223         StopCurlWatchers(IO);
224         UnlinkRSSAggregator(RSSAggr);
225         return eAbort;
226 }
227
228 eNextState RSSAggregator_TerminateDB(AsyncIO *IO)
229 {
230         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
231
232         EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
233
234
235         StopDBWatchers(&RSSAggr->IO);
236         UnlinkRSSAggregator(RSSAggr);
237         return eAbort;
238 }
239
240 eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO)
241 {
242         const char *pUrl;
243         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
244
245         pUrl = IO->ConnectMe->PlainUrl;
246         if (pUrl == NULL)
247                 pUrl = "";
248
249         EVRSSC_syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.\n", pUrl);
250
251         StopCurlWatchers(IO);
252         UnlinkRSSAggregator(RSSAggr);
253         return eAbort;
254 }
255
256 void AppendLink(StrBuf *Message,
257                 StrBuf *link,
258                 StrBuf *LinkTitle,
259                 const char *Title)
260 {
261         if (StrLength(link) > 0)
262         {
263                 StrBufAppendBufPlain(Message, HKEY("<a href=\""), 0);
264                 StrBufAppendBuf(Message, link, 0);
265                 StrBufAppendBufPlain(Message, HKEY("\">"), 0);
266                 if (StrLength(LinkTitle) > 0)
267                         StrBufAppendBuf(Message, LinkTitle, 0);
268                 else if ((Title != NULL) && !IsEmptyStr(Title))
269                         StrBufAppendBufPlain(Message, Title, -1, 0);
270                 else
271                         StrBufAppendBuf(Message, link, 0);
272                 StrBufAppendBufPlain(Message, HKEY("</a><br>\n"), 0);
273         }
274 }
275
276
277 void rss_format_item(networker_save_message *SaveMsg)
278 {
279         StrBuf *Message;
280         int msglen = 0;
281
282         if (SaveMsg->author_or_creator != NULL) {
283
284                 char *From;
285                 StrBuf *Encoded = NULL;
286                 int FromAt;
287
288                 From = html_to_ascii(ChrPtr(SaveMsg->author_or_creator),
289                                      StrLength(SaveMsg->author_or_creator),
290                                      512, 0);
291                 StrBufPlain(SaveMsg->author_or_creator, From, -1);
292                 StrBufTrim(SaveMsg->author_or_creator);
293                 free(From);
294
295                 FromAt = strchr(ChrPtr(SaveMsg->author_or_creator), '@') != NULL;
296                 if (!FromAt && StrLength (SaveMsg->author_email) > 0)
297                 {
298                         StrBufRFC2047encode(&Encoded, SaveMsg->author_or_creator);
299                         SaveMsg->Msg.cm_fields['A'] = SmashStrBuf(&Encoded);
300                         SaveMsg->Msg.cm_fields['P'] =
301                                 SmashStrBuf(&SaveMsg->author_email);
302                 }
303                 else
304                 {
305                         if (FromAt)
306                         {
307                                 SaveMsg->Msg.cm_fields['A'] =
308                                         SmashStrBuf(&SaveMsg->author_or_creator);
309                                 SaveMsg->Msg.cm_fields['P'] =
310                                         strdup(SaveMsg->Msg.cm_fields['A']);
311                         }
312                         else
313                         {
314                                 StrBufRFC2047encode(&Encoded,
315                                                     SaveMsg->author_or_creator);
316                                 SaveMsg->Msg.cm_fields['A'] =
317                                         SmashStrBuf(&Encoded);
318                                 SaveMsg->Msg.cm_fields['P'] =
319                                         strdup("rss@localhost");
320
321                         }
322                 }
323         }
324         else {
325                 SaveMsg->Msg.cm_fields['A'] = strdup("rss");
326         }
327
328         SaveMsg->Msg.cm_fields['N'] = strdup(NODENAME);
329         if (SaveMsg->title != NULL) {
330                 long len;
331                 char *Sbj;
332                 StrBuf *Encoded, *QPEncoded;
333
334                 QPEncoded = NULL;
335                 StrBufSpaceToBlank(SaveMsg->title);
336                 len = StrLength(SaveMsg->title);
337                 Sbj = html_to_ascii(ChrPtr(SaveMsg->title), len, 512, 0);
338                 len = strlen(Sbj);
339                 if ((len > 0) && (Sbj[len - 1] == '\n'))
340                 {
341                         len --;
342                         Sbj[len] = '\0';
343                 }
344                 Encoded = NewStrBufPlain(Sbj, len);
345                 free(Sbj);
346
347                 StrBufTrim(Encoded);
348                 StrBufRFC2047encode(&QPEncoded, Encoded);
349
350                 SaveMsg->Msg.cm_fields['U'] = SmashStrBuf(&QPEncoded);
351                 FreeStrBuf(&Encoded);
352         }
353         if (SaveMsg->link == NULL)
354                 SaveMsg->link = NewStrBufPlain(HKEY(""));
355
356 #if 0 /* temporarily disable shorter urls. */
357         SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] =
358                 GetShorterUrls(SaveMsg->description);
359 #endif
360
361         msglen += 1024 + StrLength(SaveMsg->link) + StrLength(SaveMsg->description) ;
362
363         Message = NewStrBufPlain(NULL, msglen);
364
365         StrBufPlain(Message, HKEY(
366                             "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n"
367                             "<html><body>\n"));
368 #if 0 /* disable shorter url for now. */
369         SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message);
370 #endif
371         StrBufAppendBuf(Message, SaveMsg->description, 0);
372         StrBufAppendBufPlain(Message, HKEY("<br><br>\n"), 0);
373
374         AppendLink(Message, SaveMsg->link, SaveMsg->linkTitle, NULL);
375         AppendLink(Message, SaveMsg->reLink, SaveMsg->reLinkTitle, "Reply to this");
376         StrBufAppendBufPlain(Message, HKEY("</body></html>\n"), 0);
377
378
379         SaveMsg->Message = Message;
380 }
381
382 eNextState RSSSaveMessage(AsyncIO *IO)
383 {
384         long len;
385         const char *Key;
386         rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data;
387
388         rss_format_item(RSSAggr->ThisMsg);
389
390         RSSAggr->ThisMsg->Msg.cm_fields['M'] =
391                 SmashStrBuf(&RSSAggr->ThisMsg->Message);
392
393         CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0);
394
395         /* write the uidl to the use table so we don't store this item again */
396
397         CheckIfAlreadySeen("RSS Item Insert", RSSAggr->ThisMsg->MsgGUID, IO->Now, 0, eWrite, IO->ID, CCID);
398
399         if (GetNextHashPos(RSSAggr->Messages,
400                            RSSAggr->Pos,
401                            &len, &Key,
402                            (void**) &RSSAggr->ThisMsg))
403                 return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
404         else
405                 return eAbort;
406 }
407
408 eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
409 {
410         const char *Key;
411         long len;
412         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
413
414         /* Find out if we've already seen this item */
415 // todo: expiry?
416 #ifndef DEBUG_RSS
417         SetRSSState(IO, eRSSUT);
418         if (CheckIfAlreadySeen("RSS Item Seen",
419                                Ctx->ThisMsg->MsgGUID,
420                                IO->Now,
421                                IO->Now - USETABLE_ANTIEXPIRE,
422                                eCheckUpdate,
423                                IO->ID, CCID)
424             != 0)
425         {
426                 /* Item has already been seen */
427                 EVRSSC_syslog(LOG_DEBUG,
428                           "%s has already been seen\n",
429                           ChrPtr(Ctx->ThisMsg->MsgGUID));
430                 SetRSSState(IO, eRSSParsing);
431
432                 if (GetNextHashPos(Ctx->Messages,
433                                    Ctx->Pos,
434                                    &len, &Key,
435                                    (void**) &Ctx->ThisMsg))
436                         return NextDBOperation(
437                                 IO,
438                                 RSS_FetchNetworkUsetableEntry);
439                 else
440                         return eAbort;
441         }
442         else
443 #endif
444         {
445                 SetRSSState(IO, eRSSParsing);
446
447                 NextDBOperation(IO, RSSSaveMessage);
448                 return eSendMore;
449         }
450         return eSendMore;
451 }
452
453 void UpdateLastKnownGood(pRSSConfig *pCfg, time_t now)
454 {
455         OneRoomNetCfg* pRNCfg;
456         begin_critical_section(S_NETCONFIGS);
457         pRNCfg = CtdlGetNetCfgForRoom (pCfg->QRnumber);
458         if (pRNCfg != NULL)
459         {
460                 RSSCfgLine *RSSCfg = (RSSCfgLine *)pRNCfg->NetConfigs[rssclient];
461
462                 while (RSSCfg != NULL)
463                 {
464                         if (RSSCfg == pCfg->pCfg)
465                                 break;
466
467                         RSSCfg = RSSCfg->next;
468                 }
469                 if (RSSCfg != NULL)
470                 {
471                         pRNCfg->changed = 1;
472                         RSSCfg->last_known_good = now;
473                 }
474         }
475
476         end_critical_section(S_NETCONFIGS);
477 }
478
479 eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
480 {
481         HashPos *it = NULL;
482         long len;
483         const char *Key;
484         pRSSConfig *pCfg;
485         u_char rawdigest[MD5_DIGEST_LEN];
486         struct MD5Context md5context;
487         StrBuf *guid;
488         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
489
490         if (IO->HttpReq.httpcode != 200)
491         {
492                 StrBuf *ErrMsg;
493                 long lens[2];
494                 const char *strs[2];
495
496                 SetRSSState(IO, eRSSFailure);
497                 ErrMsg = NewStrBuf();
498                 EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n",
499                               IO->HttpReq.httpcode);
500                 
501                 strs[0] = ChrPtr(Ctx->Url);
502                 lens[0] = StrLength(Ctx->Url);
503
504                 strs[1] = ChrPtr(Ctx->rooms);
505                 lens[1] = StrLength(Ctx->rooms);
506                 StrBufPrintf(ErrMsg,
507                              "Error while RSS-Aggregation Run of %s\n"
508                              " need a 200, got a %ld !\n"
509                              " Response text was: \n"
510                              " \n %s\n",
511                              ChrPtr(Ctx->Url),
512                              IO->HttpReq.httpcode,
513                              ChrPtr(IO->HttpReq.ReplyData));
514                 CtdlAideFPMessage(
515                         ChrPtr(ErrMsg),
516                         "RSS Aggregation run failure",
517                         2, strs, (long*) &lens,
518                         IO->Now,
519                         IO->ID, CCID);
520                 
521                 FreeStrBuf(&ErrMsg);
522                 EVRSSC_syslog(LOG_DEBUG,
523                               "RSS feed returned an invalid http status code. <%s><HTTP %ld>\n",
524                               ChrPtr(Ctx->Url),
525                               IO->HttpReq.httpcode);
526                 return eAbort;
527         }
528
529         pCfg = &Ctx->Cfg;
530
531         while (pCfg != NULL)
532         {
533                 UpdateLastKnownGood (pCfg, IO->Now);
534                 if ((Ctx->roomlist_parts > 1) && 
535                     (it == NULL))
536                 {
537                         it = GetNewHashPos(RSSFetchUrls, 0);
538                 }
539                 if (it != NULL)
540                 {
541                         void *vptr;
542                         GetNextHashPos(Ctx->OtherQRnumbers, it, &len, &Key, &vptr);
543                         pCfg = vptr;
544                 }
545                 else 
546                         pCfg = NULL;
547         }
548         DeleteHashPos (&it);
549
550         SetRSSState(IO, eRSSUT);
551
552         MD5Init(&md5context);
553
554         MD5Update(&md5context,
555                   (const unsigned char*)SKEY(IO->HttpReq.ReplyData));
556
557         MD5Update(&md5context,
558                   (const unsigned char*)SKEY(Ctx->Url));
559
560         MD5Final(rawdigest, &md5context);
561         guid = NewStrBufPlain(NULL,
562                               MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/);
563         StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN);
564         StrBufAppendBufPlain(guid, HKEY("_rssFM"), 0);
565         if (StrLength(guid) > 40)
566                 StrBufCutAt(guid, 40, NULL);
567         /* Find out if we've already seen this item */
568
569 #ifndef DEBUG_RSS
570
571         if (CheckIfAlreadySeen("RSS Whole",
572                                guid,
573                                IO->Now,
574                                IO->Now - USETABLE_ANTIEXPIRE,
575                                eCheckUpdate,
576                                IO->ID, CCID)
577             != 0)
578         {
579                 FreeStrBuf(&guid);
580
581                 EVRSSC_syslog(LOG_DEBUG, "RSS feed already seen. <%s>\n", ChrPtr(Ctx->Url));
582                 return eAbort;
583         }
584         FreeStrBuf(&guid);
585 #endif
586         SetRSSState(IO, eRSSParsing);
587         return RSSAggregator_ParseReply(IO);
588 }
589
590 eNextState RSSAggregator_FinishHttp(AsyncIO *IO)
591 {
592         return QueueDBOperation(IO, RSSAggregator_AnalyseReply);
593 }
594
595 /*
596  * Begin a feed parse
597  */
598 int rss_do_fetching(rss_aggregator *RSSAggr)
599 {
600         AsyncIO         *IO = &RSSAggr->IO;
601         rss_item *ri;
602         time_t now;
603
604         now = time(NULL);
605
606         if ((RSSAggr->next_poll != 0) && (now < RSSAggr->next_poll))
607                 return 0;
608
609         ri = (rss_item*) malloc(sizeof(rss_item));
610         memset(ri, 0, sizeof(rss_item));
611         RSSAggr->Item = ri;
612
613         if (! InitcURLIOStruct(&RSSAggr->IO,
614                                RSSAggr,
615                                "Citadel RSS Client",
616                                RSSAggregator_FinishHttp,
617                                RSSAggregator_Terminate,
618                                RSSAggregator_TerminateDB,
619                                RSSAggregator_ShutdownAbort))
620         {
621                 EVRSSCM_syslog(LOG_ALERT, "Unable to initialize libcurl.\n");
622                 return 0;
623         }
624         SetRSSState(IO, eRSSCreated);
625
626         safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host,
627                     ChrPtr(RSSAggr->Url),
628                     sizeof(((CitContext*)RSSAggr->IO.CitContext)->cs_host));
629
630         EVRSSC_syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(RSSAggr->Url));
631         ParseURL(&RSSAggr->IO.ConnectMe, RSSAggr->Url, 80);
632         CurlPrepareURL(RSSAggr->IO.ConnectMe);
633
634         SetRSSState(IO, eRSSFetching);
635         QueueCurlContext(&RSSAggr->IO);
636         return 1;
637 }
638
639 /*
640  * Scan a room's netconfig to determine whether it is requesting any RSS feeds
641  */
642 void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
643 {
644         const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
645         rss_aggregator *RSSAggr = NULL;
646         rss_aggregator *use_this_RSSAggr = NULL;
647         void *vptr;
648
649         pthread_mutex_lock(&RSSQueueMutex);
650         if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
651         {
652                 EVRSSQ_syslog(LOG_DEBUG,
653                               "rssclient: [%ld] %s already in progress.\n",
654                               qrbuf->QRnumber,
655                               qrbuf->QRname);
656                 pthread_mutex_unlock(&RSSQueueMutex);
657                 return;
658         }
659         pthread_mutex_unlock(&RSSQueueMutex);
660
661         if (server_shutting_down) return;
662
663         while (RSSCfg != NULL)
664         {
665                 pthread_mutex_lock(&RSSQueueMutex);
666                 GetHash(RSSFetchUrls,
667                         SKEY(RSSCfg->Url),
668                         &vptr);
669
670                 use_this_RSSAggr = (rss_aggregator *)vptr;
671                 if (use_this_RSSAggr != NULL)
672                 {
673                         pRSSConfig *pRSSCfg;
674
675                         StrBufAppendBufPlain(
676                                 use_this_RSSAggr->rooms,
677                                 qrbuf->QRname,
678                                 -1, 0);
679                         if (use_this_RSSAggr->roomlist_parts==1)
680                         {
681                                 use_this_RSSAggr->OtherQRnumbers
682                                         = NewHash(1, lFlathash);
683                         }
684
685                         pRSSCfg = (pRSSConfig *) malloc(sizeof(pRSSConfig));
686
687                         pRSSCfg->QRnumber = qrbuf->QRnumber;
688                         pRSSCfg->pCfg = RSSCfg;
689
690                         Put(use_this_RSSAggr->OtherQRnumbers,
691                             LKEY(qrbuf->QRnumber),
692                             pRSSCfg,
693                             NULL);
694                         use_this_RSSAggr->roomlist_parts++;
695
696                         pthread_mutex_unlock(&RSSQueueMutex);
697
698                         RSSCfg = RSSCfg->next;
699                         continue;
700                 }
701                 pthread_mutex_unlock(&RSSQueueMutex);
702
703                 RSSAggr = (rss_aggregator *) malloc(
704                         sizeof(rss_aggregator));
705
706                 memset (RSSAggr, 0, sizeof(rss_aggregator));
707                 RSSAggr->Cfg.QRnumber = qrbuf->QRnumber;
708                 RSSAggr->Cfg.pCfg = RSSCfg;
709                 RSSAggr->roomlist_parts = 1;
710                 RSSAggr->Url = NewStrBufDup(RSSCfg->Url);
711
712                 RSSAggr->ItemType = RSS_UNSET;
713
714                 RSSAggr->rooms = NewStrBufPlain(
715                         qrbuf->QRname, -1);
716
717                 pthread_mutex_lock(&RSSQueueMutex);
718
719                 Put(RSSFetchUrls,
720                     SKEY(RSSAggr->Url),
721                     RSSAggr,
722                     DeleteRssCfg);
723
724                 pthread_mutex_unlock(&RSSQueueMutex);
725                 RSSCfg = RSSCfg->next;
726         }
727 }
728
729 /*
730  * Scan for rooms that have RSS client requests configured
731  */
732 void rssclient_scan(void) {
733         int RSSRoomCount, RSSCount;
734         rss_aggregator *rptr = NULL;
735         void *vrptr = NULL;
736         HashPos *it;
737         long len;
738         const char *Key;
739         time_t now = time(NULL);
740
741         /* Run no more than once every 15 minutes. */
742         if ((now - last_run) < 900) {
743                 EVRSSQ_syslog(LOG_DEBUG,
744                               "Client: polling interval not yet reached; last run was %ldm%lds ago",
745                               ((now - last_run) / 60),
746                               ((now - last_run) % 60)
747                 );
748                 return;
749         }
750
751         /*
752          * This is a simple concurrency check to make sure only one rssclient
753          * run is done at a time.
754          */
755         pthread_mutex_lock(&RSSQueueMutex);
756         RSSCount = GetCount(RSSFetchUrls);
757         RSSRoomCount = GetCount(RSSQueueRooms);
758         pthread_mutex_unlock(&RSSQueueMutex);
759
760         if ((RSSRoomCount > 0) || (RSSCount > 0)) {
761                 EVRSSQ_syslog(LOG_DEBUG,
762                               "rssclient: concurrency check failed; %d rooms and %d url's are queued",
763                               RSSRoomCount, RSSCount
764                         );
765                 return;
766         }
767
768         become_session(&rss_CC);
769         EVRSSQM_syslog(LOG_DEBUG, "rssclient started");
770         CtdlForEachNetCfgRoom(rssclient_scan_room, NULL, rssclient);
771
772         if (GetCount(RSSFetchUrls) > 0)
773         {
774                 pthread_mutex_lock(&RSSQueueMutex);
775                 EVRSSQ_syslog(LOG_DEBUG,
776                                "rssclient starting %d Clients",
777                                GetCount(RSSFetchUrls));
778                 
779                 it = GetNewHashPos(RSSFetchUrls, 0);
780                 while (!server_shutting_down &&
781                        GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) &&
782                        (vrptr != NULL)) {
783                         rptr = (rss_aggregator *)vrptr;
784                         if (!rss_do_fetching(rptr))
785                                 UnlinkRSSAggregator(rptr);
786                 }
787                 DeleteHashPos(&it);
788                 pthread_mutex_unlock(&RSSQueueMutex);
789         }
790         else
791                 EVRSSQM_syslog(LOG_DEBUG, "Nothing to do.");
792
793         EVRSSQM_syslog(LOG_DEBUG, "rssclient ended\n");
794         return;
795 }
796
797 void rss_cleanup(void)
798 {
799         /* citthread_mutex_destroy(&RSSQueueMutex); TODO */
800         DeleteHash(&RSSFetchUrls);
801         DeleteHash(&RSSQueueRooms);
802 }
803
804 void LogDebugEnableRSSClient(const int n)
805 {
806         RSSClientDebugEnabled = n;
807 }
808
809
810 typedef struct __RSSVetoInfo {
811         StrBuf *ErrMsg;
812         time_t Now;
813         int Veto;
814 }RSSVetoInfo;
815
816 void rssclient_veto_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
817 {
818         RSSVetoInfo *Info = (RSSVetoInfo *) data;
819         const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
820
821         while (RSSCfg != NULL)
822         {
823                 if ((RSSCfg->last_known_good != 0) &&
824                     (RSSCfg->last_known_good + USETABLE_ANTIEXPIRE < Info->Now))
825                 {
826                         StrBufAppendPrintf(Info->ErrMsg,
827                                            "RSS feed not seen for a %d days:: <",
828                                            (Info->Now - RSSCfg->last_known_good) / (24 * 60 * 60));
829
830                         StrBufAppendBuf(Info->ErrMsg, RSSCfg->Url, 0);
831                         StrBufAppendBufPlain(Info->ErrMsg, HKEY(">\n"), 0);
832                 }
833                 RSSCfg = RSSCfg->next;
834         }
835 }
836
837 int RSSCheckUsetableVeto(StrBuf *ErrMsg)
838 {
839         RSSVetoInfo Info;
840
841         Info.ErrMsg = ErrMsg;
842         Info.Now = time (NULL);
843         Info.Veto = 0;
844
845         CtdlForEachNetCfgRoom(rssclient_veto_scan_room, &Info, rssclient);
846
847         return Info.Veto;;
848 }
849
850
851
852
853 void ParseRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *Line, const char *LinePos, OneRoomNetCfg *OneRNCFG)
854 {
855         RSSCfgLine *RSSCfg;
856
857         RSSCfg = (RSSCfgLine *) malloc (sizeof(RSSCfgLine));
858         RSSCfg->Url = NewStrBufPlain (NULL, StrLength (Line));
859         
860
861         StrBufExtract_NextToken(RSSCfg->Url, Line, &LinePos, '|');
862         RSSCfg->last_known_good = StrBufExtractNext_long(Line, &LinePos, '|');
863
864
865         RSSCfg->next = (RSSCfgLine *)OneRNCFG->NetConfigs[ThisOne->C];
866         OneRNCFG->NetConfigs[ThisOne->C] = (RoomNetCfgLine*) RSSCfg;
867 }
868
869 void SerializeRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *OutputBuffer, OneRoomNetCfg *RNCfg, RoomNetCfgLine *data)
870 {
871         RSSCfgLine *RSSCfg = (RSSCfgLine*) data;
872
873         StrBufAppendBufPlain(OutputBuffer, CKEY(ThisOne->Str), 0);
874         StrBufAppendBufPlain(OutputBuffer, HKEY("|"), 0);
875         StrBufAppendBufPlain(OutputBuffer, SKEY(RSSCfg->Url), 0);
876         StrBufAppendPrintf(OutputBuffer, "|%ld\n", RSSCfg->last_known_good);
877 }
878
879 void DeleteRSSClientCfgLine(const CfgLineType *ThisOne, RoomNetCfgLine **data)
880 {
881         RSSCfgLine *RSSCfg = (RSSCfgLine*) data;
882
883         FreeStrBuf(&RSSCfg->Url);
884         free(*data);
885         *data = NULL;
886 }
887
888
889 CTDL_MODULE_INIT(rssclient)
890 {
891         if (!threading)
892         {
893                 CtdlRegisterTDAPVetoHook (RSSCheckUsetableVeto, CDB_USETABLE, 0);
894
895                 CtdlREGISTERRoomCfgType(rssclient, ParseRSSClientCfgLine, 0, 1, SerializeRSSClientCfgLine, DeleteRSSClientCfgLine);
896                 pthread_mutex_init(&RSSQueueMutex, NULL);
897                 RSSQueueRooms = NewHash(1, lFlathash);
898                 RSSFetchUrls = NewHash(1, NULL);
899                 syslog(LOG_INFO, "%s\n", curl_version());
900                 CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER, PRIO_AGGR + 300);
901                 CtdlRegisterEVCleanupHook(rss_cleanup);
902                 CtdlRegisterDebugFlagHook(HKEY("rssclient"), LogDebugEnableRSSClient, &RSSClientDebugEnabled);
903         }
904         else
905         {
906                 CtdlFillSystemContext(&rss_CC, "rssclient");
907         }
908         return "rssclient";
909 }