Merge branch 'master' of ssh://git.citadel.org/appl/gitroot/citadel
[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 int rss_format_item(AsyncIO *IO, networker_save_message *SaveMsg)
278 {
279         StrBuf *Message;
280         int msglen = 0;
281
282         if (StrLength(SaveMsg->description) + 
283             StrLength(SaveMsg->link) + 
284             StrLength(SaveMsg->linkTitle) + 
285             StrLength(SaveMsg->reLink) +
286             StrLength(SaveMsg->reLinkTitle) +
287             StrLength(SaveMsg->title) == 0)
288         {
289                 EVRSSCM_syslog(LOG_INFO, "Refusing to save empty message.");
290                 return 0;
291         }
292
293         memset(&SaveMsg->Msg, 0, sizeof (struct CtdlMessage));
294
295         if (SaveMsg->author_or_creator != NULL) {
296
297                 char *From;
298                 StrBuf *Encoded = NULL;
299                 int FromAt;
300
301                 From = html_to_ascii(ChrPtr(SaveMsg->author_or_creator),
302                                      StrLength(SaveMsg->author_or_creator),
303                                      512, 0);
304                 StrBufPlain(SaveMsg->author_or_creator, From, -1);
305                 StrBufTrim(SaveMsg->author_or_creator);
306                 free(From);
307
308                 FromAt = strchr(ChrPtr(SaveMsg->author_or_creator), '@') != NULL;
309                 if (!FromAt && StrLength (SaveMsg->author_email) > 0)
310                 {
311                         StrBufRFC2047encode(&Encoded, SaveMsg->author_or_creator);
312                         CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &Encoded);
313                         CM_SetAsFieldSB(&SaveMsg->Msg, eMessagePath, &SaveMsg->author_email);
314                 }
315                 else
316                 {
317                         if (FromAt)
318                         {
319                                 CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &SaveMsg->author_or_creator);
320                                 CM_CopyField(&SaveMsg->Msg, eMessagePath, eAuthor);
321                         }
322                         else
323                         {
324                                 StrBufRFC2047encode(&Encoded,
325                                                     SaveMsg->author_or_creator);
326                                 CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &Encoded);
327                                 CM_SetField(&SaveMsg->Msg, eMessagePath, HKEY("rss@localhost"));
328
329                         }
330                 }
331         }
332         else {
333                 CM_SetField(&SaveMsg->Msg, eAuthor, HKEY("rss"));
334         }
335
336         CM_SetField(&SaveMsg->Msg, eNodeName, NODENAME, strlen(NODENAME));
337         if (SaveMsg->title != NULL) {
338                 long len;
339                 char *Sbj;
340                 StrBuf *Encoded, *QPEncoded;
341
342                 QPEncoded = NULL;
343                 StrBufSpaceToBlank(SaveMsg->title);
344                 len = StrLength(SaveMsg->title);
345                 Sbj = html_to_ascii(ChrPtr(SaveMsg->title), len, 512, 0);
346                 len = strlen(Sbj);
347                 if ((len > 0) && (Sbj[len - 1] == '\n'))
348                 {
349                         len --;
350                         Sbj[len] = '\0';
351                 }
352                 Encoded = NewStrBufPlain(Sbj, len);
353                 free(Sbj);
354
355                 StrBufTrim(Encoded);
356                 StrBufRFC2047encode(&QPEncoded, Encoded);
357
358                 CM_SetAsFieldSB(&SaveMsg->Msg, eMsgSubject, &QPEncoded);
359                 FreeStrBuf(&Encoded);
360         }
361         if (SaveMsg->link == NULL)
362                 SaveMsg->link = NewStrBufPlain(HKEY(""));
363
364 #if 0 /* temporarily disable shorter urls. */
365         SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] =
366                 GetShorterUrls(SaveMsg->description);
367 #endif
368
369         msglen += 1024 + StrLength(SaveMsg->link) + StrLength(SaveMsg->description) ;
370
371         Message = NewStrBufPlain(NULL, msglen);
372
373         StrBufPlain(Message, HKEY(
374                             "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n"
375                             "<html><body>\n"));
376 #if 0 /* disable shorter url for now. */
377         SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message);
378 #endif
379         StrBufAppendBuf(Message, SaveMsg->description, 0);
380         StrBufAppendBufPlain(Message, HKEY("<br><br>\n"), 0);
381
382         AppendLink(Message, SaveMsg->link, SaveMsg->linkTitle, NULL);
383         AppendLink(Message, SaveMsg->reLink, SaveMsg->reLinkTitle, "Reply to this");
384         StrBufAppendBufPlain(Message, HKEY("</body></html>\n"), 0);
385
386         SaveMsg->Message = Message;
387         return 1;
388 }
389
390 eNextState RSSSaveMessage(AsyncIO *IO)
391 {
392         long len;
393         const char *Key;
394         rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data;
395
396         if (rss_format_item(IO, RSSAggr->ThisMsg))
397         {
398                 CM_SetAsFieldSB(&RSSAggr->ThisMsg->Msg, eMesageText,
399                                        &RSSAggr->ThisMsg->Message);
400
401                 CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0);
402                 
403                 /* write the uidl to the use table so we don't store this item again */
404                 
405                 CheckIfAlreadySeen("RSS Item Insert", RSSAggr->ThisMsg->MsgGUID, IO->Now, 0, eWrite, CCID, IO->ID);
406         }
407
408         if (GetNextHashPos(RSSAggr->Messages,
409                            RSSAggr->Pos,
410                            &len, &Key,
411                            (void**) &RSSAggr->ThisMsg))
412                 return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
413         else
414                 return eAbort;
415 }
416
417 eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
418 {
419 #ifndef DEBUG_RSS
420         const char *Key;
421         long len;
422         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
423
424         /* Find out if we've already seen this item */
425 // todo: expiry?
426         SetRSSState(IO, eRSSUT);
427         if (CheckIfAlreadySeen("RSS Item Seen",
428                                Ctx->ThisMsg->MsgGUID,
429                                IO->Now,
430                                IO->Now - USETABLE_ANTIEXPIRE_HIRES,
431                                eCheckUpdate,
432                                CCID, IO->ID)
433             != 0)
434         {
435                 /* Item has already been seen */
436                 EVRSSC_syslog(LOG_DEBUG,
437                           "%s has already been seen\n",
438                           ChrPtr(Ctx->ThisMsg->MsgGUID));
439                 SetRSSState(IO, eRSSParsing);
440
441                 if (GetNextHashPos(Ctx->Messages,
442                                    Ctx->Pos,
443                                    &len, &Key,
444                                    (void**) &Ctx->ThisMsg))
445                         return NextDBOperation(
446                                 IO,
447                                 RSS_FetchNetworkUsetableEntry);
448                 else
449                         return eAbort;
450         }
451         else
452 #endif
453         {
454                 SetRSSState(IO, eRSSParsing);
455
456                 NextDBOperation(IO, RSSSaveMessage);
457                 return eSendMore;
458         }
459         return eSendMore;
460 }
461
462 void UpdateLastKnownGood(pRSSConfig *pCfg, time_t now)
463 {
464         OneRoomNetCfg* pRNCfg;
465         begin_critical_section(S_NETCONFIGS);
466         pRNCfg = CtdlGetNetCfgForRoom (pCfg->QRnumber);
467         if (pRNCfg != NULL)
468         {
469                 RSSCfgLine *RSSCfg = (RSSCfgLine *)pRNCfg->NetConfigs[rssclient];
470
471                 while (RSSCfg != NULL)
472                 {
473                         if (RSSCfg == pCfg->pCfg)
474                                 break;
475
476                         RSSCfg = RSSCfg->next;
477                 }
478                 if (RSSCfg != NULL)
479                 {
480                         pRNCfg->changed = 1;
481                         RSSCfg->last_known_good = now;
482                 }
483         }
484
485         end_critical_section(S_NETCONFIGS);
486 }
487
488 eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
489 {
490         HashPos *it = NULL;
491         long len;
492         const char *Key;
493         pRSSConfig *pCfg;
494         u_char rawdigest[MD5_DIGEST_LEN];
495         struct MD5Context md5context;
496         StrBuf *guid;
497         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
498
499         if (IO->HttpReq.httpcode != 200)
500         {
501                 StrBuf *ErrMsg;
502                 long lens[2];
503                 const char *strs[2];
504
505                 SetRSSState(IO, eRSSFailure);
506                 ErrMsg = NewStrBuf();
507                 EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n",
508                               IO->HttpReq.httpcode);
509                 
510                 strs[0] = ChrPtr(Ctx->Url);
511                 lens[0] = StrLength(Ctx->Url);
512
513                 strs[1] = ChrPtr(Ctx->rooms);
514                 lens[1] = StrLength(Ctx->rooms);
515
516                 if (IO->HttpReq.CurlError == NULL)
517                         IO->HttpReq.CurlError = "";
518
519                 StrBufPrintf(ErrMsg,
520                              "Error while RSS-Aggregation Run of %s\n"
521                              " need a 200, got a %ld !\n"
522                              " Curl Error message: \n%s / %s\n"
523                              " Response text was: \n"
524                              " \n %s\n",
525                              ChrPtr(Ctx->Url),
526                              IO->HttpReq.httpcode,
527                              IO->HttpReq.errdesc,
528                              IO->HttpReq.CurlError,
529                              ChrPtr(IO->HttpReq.ReplyData)
530                         );
531
532                 CtdlAideFPMessage(
533                         ChrPtr(ErrMsg),
534                         "RSS Aggregation run failure",
535                         2, strs, (long*) &lens,
536                         IO->Now,
537                         IO->ID, CCID);
538                 
539                 FreeStrBuf(&ErrMsg);
540                 EVRSSC_syslog(LOG_DEBUG,
541                               "RSS feed returned an invalid http status code. <%s><HTTP %ld>\n",
542                               ChrPtr(Ctx->Url),
543                               IO->HttpReq.httpcode);
544                 return eAbort;
545         }
546
547         pCfg = &Ctx->Cfg;
548
549         while (pCfg != NULL)
550         {
551                 UpdateLastKnownGood (pCfg, IO->Now);
552                 if ((Ctx->roomlist_parts > 1) && 
553                     (it == NULL))
554                 {
555                         it = GetNewHashPos(RSSFetchUrls, 0);
556                 }
557                 if (it != NULL)
558                 {
559                         void *vptr;
560                         if (GetNextHashPos(Ctx->OtherQRnumbers, it, &len, &Key, &vptr))
561                                 pCfg = vptr;
562                         else
563                                 pCfg = NULL;
564                 }
565                 else 
566                         pCfg = NULL;
567         }
568         DeleteHashPos (&it);
569
570         SetRSSState(IO, eRSSUT);
571
572         MD5Init(&md5context);
573
574         MD5Update(&md5context,
575                   (const unsigned char*)SKEY(IO->HttpReq.ReplyData));
576
577         MD5Update(&md5context,
578                   (const unsigned char*)SKEY(Ctx->Url));
579
580         MD5Final(rawdigest, &md5context);
581         guid = NewStrBufPlain(NULL,
582                               MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/);
583         StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN);
584         StrBufAppendBufPlain(guid, HKEY("_rssFM"), 0);
585         if (StrLength(guid) > 40)
586                 StrBufCutAt(guid, 40, NULL);
587         /* Find out if we've already seen this item */
588
589 #ifndef DEBUG_RSS
590
591         if (CheckIfAlreadySeen("RSS Whole",
592                                guid,
593                                IO->Now,
594                                IO->Now - USETABLE_ANTIEXPIRE,
595                                eCheckUpdate,
596                                CCID, IO->ID)
597             != 0)
598         {
599                 FreeStrBuf(&guid);
600
601                 EVRSSC_syslog(LOG_DEBUG, "RSS feed already seen. <%s>\n", ChrPtr(Ctx->Url));
602                 return eAbort;
603         }
604         FreeStrBuf(&guid);
605 #endif
606         SetRSSState(IO, eRSSParsing);
607         return RSSAggregator_ParseReply(IO);
608 }
609
610 eNextState RSSAggregator_FinishHttp(AsyncIO *IO)
611 {
612         return CurlQueueDBOperation(IO, RSSAggregator_AnalyseReply);
613 }
614
615 /*
616  * Begin a feed parse
617  */
618 int rss_do_fetching(rss_aggregator *RSSAggr)
619 {
620         AsyncIO         *IO = &RSSAggr->IO;
621         rss_item *ri;
622         time_t now;
623
624         now = time(NULL);
625
626         if ((RSSAggr->next_poll != 0) && (now < RSSAggr->next_poll))
627                 return 0;
628
629         ri = (rss_item*) malloc(sizeof(rss_item));
630         memset(ri, 0, sizeof(rss_item));
631         RSSAggr->Item = ri;
632
633         if (! InitcURLIOStruct(&RSSAggr->IO,
634                                RSSAggr,
635                                "Citadel RSS Client",
636                                RSSAggregator_FinishHttp,
637                                RSSAggregator_Terminate,
638                                RSSAggregator_TerminateDB,
639                                RSSAggregator_ShutdownAbort))
640         {
641                 EVRSSCM_syslog(LOG_ALERT, "Unable to initialize libcurl.\n");
642                 return 0;
643         }
644         SetRSSState(IO, eRSSCreated);
645
646         safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host,
647                     ChrPtr(RSSAggr->Url),
648                     sizeof(((CitContext*)RSSAggr->IO.CitContext)->cs_host));
649
650         EVRSSC_syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(RSSAggr->Url));
651         ParseURL(&RSSAggr->IO.ConnectMe, RSSAggr->Url, 80);
652         CurlPrepareURL(RSSAggr->IO.ConnectMe);
653
654         SetRSSState(IO, eRSSFetching);
655         QueueCurlContext(&RSSAggr->IO);
656         return 1;
657 }
658
659 /*
660  * Scan a room's netconfig to determine whether it is requesting any RSS feeds
661  */
662 void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
663 {
664         const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
665         rss_aggregator *RSSAggr = NULL;
666         rss_aggregator *use_this_RSSAggr = NULL;
667         void *vptr;
668
669         pthread_mutex_lock(&RSSQueueMutex);
670         if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
671         {
672                 EVRSSQ_syslog(LOG_DEBUG,
673                               "rssclient: [%ld] %s already in progress.\n",
674                               qrbuf->QRnumber,
675                               qrbuf->QRname);
676                 pthread_mutex_unlock(&RSSQueueMutex);
677                 return;
678         }
679         pthread_mutex_unlock(&RSSQueueMutex);
680
681         if (server_shutting_down) return;
682
683         while (RSSCfg != NULL)
684         {
685                 pthread_mutex_lock(&RSSQueueMutex);
686                 GetHash(RSSFetchUrls,
687                         SKEY(RSSCfg->Url),
688                         &vptr);
689
690                 use_this_RSSAggr = (rss_aggregator *)vptr;
691                 if (use_this_RSSAggr != NULL)
692                 {
693                         pRSSConfig *pRSSCfg;
694
695                         StrBufAppendBufPlain(
696                                 use_this_RSSAggr->rooms,
697                                 qrbuf->QRname,
698                                 -1, 0);
699                         if (use_this_RSSAggr->roomlist_parts==1)
700                         {
701                                 use_this_RSSAggr->OtherQRnumbers
702                                         = NewHash(1, lFlathash);
703                         }
704
705                         pRSSCfg = (pRSSConfig *) malloc(sizeof(pRSSConfig));
706
707                         pRSSCfg->QRnumber = qrbuf->QRnumber;
708                         pRSSCfg->pCfg = RSSCfg;
709
710                         Put(use_this_RSSAggr->OtherQRnumbers,
711                             LKEY(qrbuf->QRnumber),
712                             pRSSCfg,
713                             NULL);
714                         use_this_RSSAggr->roomlist_parts++;
715
716                         pthread_mutex_unlock(&RSSQueueMutex);
717
718                         RSSCfg = RSSCfg->next;
719                         continue;
720                 }
721                 pthread_mutex_unlock(&RSSQueueMutex);
722
723                 RSSAggr = (rss_aggregator *) malloc(
724                         sizeof(rss_aggregator));
725
726                 memset (RSSAggr, 0, sizeof(rss_aggregator));
727                 RSSAggr->Cfg.QRnumber = qrbuf->QRnumber;
728                 RSSAggr->Cfg.pCfg = RSSCfg;
729                 RSSAggr->roomlist_parts = 1;
730                 RSSAggr->Url = NewStrBufDup(RSSCfg->Url);
731
732                 RSSAggr->ItemType = RSS_UNSET;
733
734                 RSSAggr->rooms = NewStrBufPlain(
735                         qrbuf->QRname, -1);
736
737                 pthread_mutex_lock(&RSSQueueMutex);
738
739                 Put(RSSFetchUrls,
740                     SKEY(RSSAggr->Url),
741                     RSSAggr,
742                     DeleteRssCfg);
743
744                 pthread_mutex_unlock(&RSSQueueMutex);
745                 RSSCfg = RSSCfg->next;
746         }
747 }
748
749 /*
750  * Scan for rooms that have RSS client requests configured
751  */
752 void rssclient_scan(void) {
753         int RSSRoomCount, RSSCount;
754         rss_aggregator *rptr = NULL;
755         void *vrptr = NULL;
756         HashPos *it;
757         long len;
758         const char *Key;
759         time_t now = time(NULL);
760
761         /* Run no more than once every 15 minutes. */
762         if ((now - last_run) < 900) {
763                 EVRSSQ_syslog(LOG_DEBUG,
764                               "Client: polling interval not yet reached; last run was %ldm%lds ago",
765                               ((now - last_run) / 60),
766                               ((now - last_run) % 60)
767                 );
768                 return;
769         }
770
771         /*
772          * This is a simple concurrency check to make sure only one rssclient
773          * run is done at a time.
774          */
775         pthread_mutex_lock(&RSSQueueMutex);
776         RSSCount = GetCount(RSSFetchUrls);
777         RSSRoomCount = GetCount(RSSQueueRooms);
778         pthread_mutex_unlock(&RSSQueueMutex);
779
780         if ((RSSRoomCount > 0) || (RSSCount > 0)) {
781                 EVRSSQ_syslog(LOG_DEBUG,
782                               "rssclient: concurrency check failed; %d rooms and %d url's are queued",
783                               RSSRoomCount, RSSCount
784                         );
785                 return;
786         }
787
788         become_session(&rss_CC);
789         EVRSSQM_syslog(LOG_DEBUG, "rssclient started");
790         CtdlForEachNetCfgRoom(rssclient_scan_room, NULL, rssclient);
791
792         if (GetCount(RSSFetchUrls) > 0)
793         {
794                 pthread_mutex_lock(&RSSQueueMutex);
795                 EVRSSQ_syslog(LOG_DEBUG,
796                                "rssclient starting %d Clients",
797                                GetCount(RSSFetchUrls));
798                 
799                 it = GetNewHashPos(RSSFetchUrls, 0);
800                 while (!server_shutting_down &&
801                        GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) &&
802                        (vrptr != NULL)) {
803                         rptr = (rss_aggregator *)vrptr;
804                         if (!rss_do_fetching(rptr))
805                                 UnlinkRSSAggregator(rptr);
806                 }
807                 DeleteHashPos(&it);
808                 pthread_mutex_unlock(&RSSQueueMutex);
809         }
810         else
811                 EVRSSQM_syslog(LOG_DEBUG, "Nothing to do.");
812
813         EVRSSQM_syslog(LOG_DEBUG, "rssclient ended\n");
814         return;
815 }
816
817 void rss_cleanup(void)
818 {
819         /* citthread_mutex_destroy(&RSSQueueMutex); TODO */
820         DeleteHash(&RSSFetchUrls);
821         DeleteHash(&RSSQueueRooms);
822 }
823
824 void LogDebugEnableRSSClient(const int n)
825 {
826         RSSClientDebugEnabled = n;
827 }
828
829
830 typedef struct __RSSVetoInfo {
831         StrBuf *ErrMsg;
832         time_t Now;
833         int Veto;
834 }RSSVetoInfo;
835
836 void rssclient_veto_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
837 {
838         RSSVetoInfo *Info = (RSSVetoInfo *) data;
839         const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
840
841         while (RSSCfg != NULL)
842         {
843                 if ((RSSCfg->last_known_good != 0) &&
844                     (RSSCfg->last_known_good + USETABLE_ANTIEXPIRE < Info->Now))
845                 {
846                         StrBufAppendPrintf(Info->ErrMsg,
847                                            "RSS feed not seen for a %d days:: <",
848                                            (Info->Now - RSSCfg->last_known_good) / (24 * 60 * 60));
849
850                         StrBufAppendBuf(Info->ErrMsg, RSSCfg->Url, 0);
851                         StrBufAppendBufPlain(Info->ErrMsg, HKEY(">\n"), 0);
852                 }
853                 RSSCfg = RSSCfg->next;
854         }
855 }
856
857 int RSSCheckUsetableVeto(StrBuf *ErrMsg)
858 {
859         RSSVetoInfo Info;
860
861         Info.ErrMsg = ErrMsg;
862         Info.Now = time (NULL);
863         Info.Veto = 0;
864
865         CtdlForEachNetCfgRoom(rssclient_veto_scan_room, &Info, rssclient);
866
867         return Info.Veto;;
868 }
869
870
871
872
873 void ParseRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *Line, const char *LinePos, OneRoomNetCfg *OneRNCFG)
874 {
875         RSSCfgLine *RSSCfg;
876
877         RSSCfg = (RSSCfgLine *) malloc (sizeof(RSSCfgLine));
878         RSSCfg->Url = NewStrBufPlain (NULL, StrLength (Line));
879         
880
881         StrBufExtract_NextToken(RSSCfg->Url, Line, &LinePos, '|');
882         RSSCfg->last_known_good = StrBufExtractNext_long(Line, &LinePos, '|');
883
884
885         RSSCfg->next = (RSSCfgLine *)OneRNCFG->NetConfigs[ThisOne->C];
886         OneRNCFG->NetConfigs[ThisOne->C] = (RoomNetCfgLine*) RSSCfg;
887 }
888
889 void SerializeRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *OutputBuffer, OneRoomNetCfg *RNCfg, RoomNetCfgLine *data)
890 {
891         RSSCfgLine *RSSCfg = (RSSCfgLine*) data;
892
893         StrBufAppendBufPlain(OutputBuffer, CKEY(ThisOne->Str), 0);
894         StrBufAppendBufPlain(OutputBuffer, HKEY("|"), 0);
895         StrBufAppendBufPlain(OutputBuffer, SKEY(RSSCfg->Url), 0);
896         StrBufAppendPrintf(OutputBuffer, "|%ld\n", RSSCfg->last_known_good);
897 }
898
899 void DeleteRSSClientCfgLine(const CfgLineType *ThisOne, RoomNetCfgLine **data)
900 {
901         RSSCfgLine *RSSCfg = (RSSCfgLine*) *data;
902
903         FreeStrBuf(&RSSCfg->Url);
904         free(*data);
905         *data = NULL;
906 }
907
908
909 CTDL_MODULE_INIT(rssclient)
910 {
911         if (!threading)
912         {
913                 CtdlRegisterTDAPVetoHook (RSSCheckUsetableVeto, CDB_USETABLE, 0);
914
915                 CtdlREGISTERRoomCfgType(rssclient, ParseRSSClientCfgLine, 0, 1, SerializeRSSClientCfgLine, DeleteRSSClientCfgLine);
916                 pthread_mutex_init(&RSSQueueMutex, NULL);
917                 RSSQueueRooms = NewHash(1, lFlathash);
918                 RSSFetchUrls = NewHash(1, NULL);
919                 syslog(LOG_INFO, "%s\n", curl_version());
920                 CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER, PRIO_AGGR + 300);
921                 CtdlRegisterEVCleanupHook(rss_cleanup);
922                 CtdlRegisterDebugFlagHook(HKEY("rssclient"), LogDebugEnableRSSClient, &RSSClientDebugEnabled);
923         }
924         else
925         {
926                 CtdlFillSystemContext(&rss_CC, "rssclient");
927         }
928         return "rssclient";
929 }