Work on evented RSS client & libev+libcurl integration
[citadel.git] / citadel / modules / rssclient / serv_rssclient.c
1 /*
2  * Bring external RSS feeds into rooms.
3  *
4  * Copyright (c) 2007-2010 by the citadel.org team
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation; either version 3 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
19  */
20
21 #include <stdlib.h>
22 #include <unistd.h>
23 #include <stdio.h>
24
25 #if TIME_WITH_SYS_TIME
26 # include <sys/time.h>
27 # include <time.h>
28 #else
29 # if HAVE_SYS_TIME_H
30 #  include <sys/time.h>
31 # else
32 #  include <time.h>
33 # endif
34 #endif
35
36 #include <ctype.h>
37 #include <string.h>
38 #include <errno.h>
39 #include <sys/types.h>
40 #include <sys/stat.h>
41 #include <expat.h>
42 #include <curl/curl.h>
43 #include <libcitadel.h>
44 #include "citadel.h"
45 #include "server.h"
46 #include "citserver.h"
47 #include "support.h"
48 #include "config.h"
49 #include "threads.h"
50 #include "ctdl_module.h"
51 #include "msgbase.h"
52 #include "parsedate.h"
53 #include "database.h"
54 #include "citadel_dirs.h"
55 #include "md5.h"
56 #include "context.h"
57 #include "event_client.h"
58 #include "rss_atom_parser.h"
59
60
61 #define TMP_MSGDATA 0xFF
62 #define TMP_SHORTER_URL_OFFSET 0xFE
63 #define TMP_SHORTER_URLS 0xFD
64
65 time_t last_run = 0L;
66
67 pthread_mutex_t RSSQueueMutex; /* locks the access to the following vars: */
68 HashList *RSSQueueRooms = NULL; /* rss_room_counter */
69 HashList *RSSFetchUrls = NULL; /* -> rss_aggregator; ->RefCount access to be locked too. */
70
71 eNextState RSSAggregatorTerminate(AsyncIO *IO);
72
73 struct CitContext rss_CC;
74
75 struct rssnetcfg *rnclist = NULL;
76 void AppendLink(StrBuf *Message, StrBuf *link, StrBuf *LinkTitle, const char *Title)
77 {
78         if (StrLength(link) > 0)
79         {
80                 StrBufAppendBufPlain(Message, HKEY("<a href=\""), 0);
81                 StrBufAppendBuf(Message, link, 0);
82                 StrBufAppendBufPlain(Message, HKEY("\">"), 0);
83                 if (StrLength(LinkTitle) > 0)
84                         StrBufAppendBuf(Message, LinkTitle, 0);
85                 else if ((Title != NULL) && !IsEmptyStr(Title))
86                         StrBufAppendBufPlain(Message, Title, -1, 0);
87                 else
88                         StrBufAppendBuf(Message, link, 0);
89                 StrBufAppendBufPlain(Message, HKEY("</a><br>\n"), 0);
90         }
91 }
92
93
94 void DeleteRoomReference(long QRnumber)
95 {
96         HashPos *At;
97         long HKLen;
98         const char *HK;
99         void *vData = NULL;
100         rss_room_counter *pRoomC;
101
102         At = GetNewHashPos(RSSQueueRooms, 0);
103
104         GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At);
105         GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData);
106         if (vData != NULL)
107         {
108                 pRoomC = (rss_room_counter *) vData;
109                 pRoomC->count --;
110                 if (pRoomC->count == 0)
111                         DeleteEntryFromHash(RSSQueueRooms, At);
112         }
113         DeleteHashPos(&At);
114 }
115
116 void UnlinkRooms(rss_aggregator *Cfg)
117 {
118         
119         DeleteRoomReference(Cfg->QRnumber);
120         if (Cfg->OtherQRnumbers != NULL)
121         {
122                 long HKLen;
123                 const char *HK;
124                 HashPos *At;
125                 void *vData;
126
127                 At = GetNewHashPos(Cfg->OtherQRnumbers, 0);
128                 while (GetNextHashPos(Cfg->OtherQRnumbers, At, &HKLen, &HK, &vData) && 
129                        (vData != NULL))
130                 {
131                         long *lData = (long*) vData;
132                         DeleteRoomReference(*lData);
133                 }
134 /*
135                 if (server_shutting_down)
136                         break; / * TODO */
137
138                 DeleteHashPos(&At);
139         }
140 }
141
142 void UnlinkRSSAggregator(rss_aggregator *Cfg)
143 {
144         HashPos *At;
145
146         UnlinkRooms(Cfg);
147
148         At = GetNewHashPos(RSSFetchUrls, 0);
149         if (GetHashPosFromKey(RSSFetchUrls, SKEY(Cfg->Url), At) == 0)
150         {
151                 DeleteEntryFromHash(RSSFetchUrls, At);
152         }
153         DeleteHashPos(&At);
154         last_run = time(NULL);
155 }
156 /*
157 eNextState FreeNetworkSaveMessage (AsyncIO *IO)
158 {
159         networker_save_message *Ctx = (networker_save_message *) IO->Data;
160
161         pthread_mutex_lock(&RSSQueueMutex);
162         Ctx->Cfg->RefCount --;
163
164         if (Ctx->Cfg->RefCount == 0)
165         {
166                 UnlinkRSSAggregator(Ctx->Cfg);
167
168         }
169         pthread_mutex_unlock(&RSSQueueMutex);
170
171         CtdlFreeMessage(Ctx->Msg);
172         free_recipients(Ctx->recp);
173         FreeStrBuf(&Ctx->Message);
174         FreeStrBuf(&Ctx->MsgGUID);
175         ((struct CitContext*)IO->CitContext)->state = CON_IDLE;
176         ((struct CitContext*)IO->CitContext)->kill_me = 1;
177         free(Ctx);
178         last_run = time(NULL);
179         return eAbort;
180 }
181 */
182 void FreeNetworkSaveMessage (void *vMsg)
183 {
184         networker_save_message *Msg = (networker_save_message *) vMsg;
185
186         CtdlFreeMessage(Msg->Msg);
187         FreeStrBuf(&Msg->Message);
188         FreeStrBuf(&Msg->MsgGUID);
189         free(Msg);
190 }
191
192 eNextState AbortNetworkSaveMessage (AsyncIO *IO)
193 {
194         return eAbort; ///TODO
195 }
196
197 eNextState RSSSaveMessage(AsyncIO *IO)
198 {
199         long len;
200         const char *Key;
201         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
202
203         Ctx->ThisMsg->Msg->cm_fields['M'] = SmashStrBuf(&Ctx->ThisMsg->Message);
204
205         CtdlSubmitMsg(Ctx->ThisMsg->Msg, &Ctx->recp, NULL, 0);
206
207         /* write the uidl to the use table so we don't store this item again */
208         cdb_store(CDB_USETABLE, SKEY(Ctx->ThisMsg->MsgGUID), &Ctx->ThisMsg->ut, sizeof(struct UseTable) );
209         
210         if (GetNextHashPos(Ctx->Messages, Ctx->Pos, &len, &Key, (void**) &Ctx->ThisMsg))
211                 return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
212         else
213                 return eAbort;
214 }
215
216 eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
217 {
218         const char *Key;
219         long len;
220         struct cdbdata *cdbut;
221         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
222
223
224         /* Find out if we've already seen this item */
225         strcpy(Ctx->ThisMsg->ut.ut_msgid, ChrPtr(Ctx->ThisMsg->MsgGUID)); /// TODO
226         Ctx->ThisMsg->ut.ut_timestamp = time(NULL);
227
228         cdbut = cdb_fetch(CDB_USETABLE, SKEY(Ctx->ThisMsg->MsgGUID));
229 #ifndef DEBUG_RSS
230         if (cdbut != NULL) {
231                 /* Item has already been seen */
232                 EV_syslog(LOG_DEBUG, "%s has already been seen\n", ChrPtr(Ctx->ThisMsg->MsgGUID));
233                 cdb_free(cdbut);
234
235                 /* rewrite the record anyway, to update the timestamp */
236                 cdb_store(CDB_USETABLE, 
237                           SKEY(Ctx->ThisMsg->MsgGUID), 
238                           &Ctx->ThisMsg->ut, sizeof(struct UseTable) );
239
240                 if (GetNextHashPos(Ctx->Messages, Ctx->Pos, &len, &Key, (void**) &Ctx->ThisMsg))
241                         return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
242                 else
243                         return eAbort;
244         }
245         else
246 #endif
247         {
248                 NextDBOperation(IO, RSSSaveMessage);
249                 return eSendMore;
250         }
251 }
252 /*
253 void RSSAddSaveMessage(struct CtdlMessage *Msg, struct recptypes *recp, StrBuf *MsgGUID, StrBuf *MessageBody, rss_aggregat *Cfg)
254 {
255         networker_save_message *Ctx;
256
257         pthread_mutex_lock(&RSSQueueMutex);
258         Cfg->RefCount ++;
259         pthread_mutex_unlock(&RSSQueueMutex);
260
261
262         Ctx = (networker_save_message *) malloc(sizeof(networker_save_message));
263         memset(Ctx, 0, sizeof(networker_save_message));
264         
265         Ctx->MsgGUID = MsgGUID;
266         Ctx->Message = MessageBody;
267         Ctx->Msg = Msg;
268         Ctx->Cfg = Cfg;
269         Ctx->recp = recp;
270         Ctx->IO.Data = Ctx;
271         Ctx->IO.CitContext = CloneContext(&rss_CC);
272         Ctx->IO.Terminate = FreeNetworkSaveMessage;
273         Ctx->IO.ShutdownAbort = AbortNetworkSaveMessage;
274         QueueDBOperation(&Ctx->IO, RSS_FetchNetworkUsetableEntry);
275 }
276 */
277
278 /*
279  * Commit a fetched and parsed RSS item to disk
280  */
281 void rss_save_item(rss_item *ri, rss_aggregator *Cfg)
282 {
283
284         struct MD5Context md5context;
285         u_char rawdigest[MD5_DIGEST_LEN];
286         struct CtdlMessage *msg;
287         int msglen = 0;
288         StrBuf *Message;
289         StrBuf *guid;
290         AsyncIO *IO = &Cfg->IO;
291
292         int n;
293    
294         /* Construct a GUID to use in the S_USETABLE table.
295          * If one is not present in the item itself, make one up.
296          */
297         if (ri->guid != NULL) {
298                 StrBufSpaceToBlank(ri->guid);
299                 StrBufTrim(ri->guid);
300                 guid = NewStrBufPlain(HKEY("rss/"));
301                 StrBufAppendBuf(guid, ri->guid, 0);
302         }
303         else {
304                 MD5Init(&md5context);
305                 if (ri->title != NULL) {
306                         MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->title), StrLength(ri->title));
307                 }
308                 if (ri->link != NULL) {
309                         MD5Update(&md5context, (const unsigned char*)ChrPtr(ri->link), StrLength(ri->link));
310                 }
311                 MD5Final(rawdigest, &md5context);
312                 guid = NewStrBufPlain(NULL, MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/);
313                 StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN);
314                 StrBufAppendBufPlain(guid, HKEY("_rss2ctdl"), 0);
315         }
316
317         /* translate Item into message. */
318         EVM_syslog(LOG_DEBUG, "RSS: translating item...\n");
319         if (ri->description == NULL) ri->description = NewStrBufPlain(HKEY(""));
320         StrBufSpaceToBlank(ri->description);
321         msg = malloc(sizeof(struct CtdlMessage));
322         memset(msg, 0, sizeof(struct CtdlMessage));
323         msg->cm_magic = CTDLMESSAGE_MAGIC;
324         msg->cm_anon_type = MES_NORMAL;
325         msg->cm_format_type = FMT_RFC822;
326
327         if (ri->guid != NULL) {
328                 msg->cm_fields['E'] = strdup(ChrPtr(ri->guid));
329         }
330
331         if (ri->author_or_creator != NULL) {
332                 char *From;
333                 StrBuf *Encoded = NULL;
334                 int FromAt;
335                         
336                 From = html_to_ascii(ChrPtr(ri->author_or_creator),
337                                      StrLength(ri->author_or_creator), 
338                                      512, 0);
339                 StrBufPlain(ri->author_or_creator, From, -1);
340                 StrBufTrim(ri->author_or_creator);
341                 free(From);
342
343                 FromAt = strchr(ChrPtr(ri->author_or_creator), '@') != NULL;
344                 if (!FromAt && StrLength (ri->author_email) > 0)
345                 {
346                         StrBufRFC2047encode(&Encoded, ri->author_or_creator);
347                         msg->cm_fields['A'] = SmashStrBuf(&Encoded);
348                         msg->cm_fields['P'] = SmashStrBuf(&ri->author_email);
349                 }
350                 else
351                 {
352                         if (FromAt)
353                         {
354                                 msg->cm_fields['A'] = SmashStrBuf(&ri->author_or_creator);
355                                 msg->cm_fields['P'] = strdup(msg->cm_fields['A']);
356                         }
357                         else 
358                         {
359                                 StrBufRFC2047encode(&Encoded, ri->author_or_creator);
360                                 msg->cm_fields['A'] = SmashStrBuf(&Encoded);
361                                 msg->cm_fields['P'] = strdup("rss@localhost");
362
363                         }
364                         if (ri->pubdate <= 0) {
365                                 ri->pubdate = time(NULL);
366                         }
367                 }
368         }
369         else {
370                 msg->cm_fields['A'] = strdup("rss");
371         }
372
373         msg->cm_fields['N'] = strdup(NODENAME);
374         if (ri->title != NULL) {
375                 long len;
376                 char *Sbj;
377                 StrBuf *Encoded, *QPEncoded;
378
379                 QPEncoded = NULL;
380                 StrBufSpaceToBlank(ri->title);
381                 len = StrLength(ri->title);
382                 Sbj = html_to_ascii(ChrPtr(ri->title), len, 512, 0);
383                 len = strlen(Sbj);
384                 if (Sbj[len - 1] == '\n')
385                 {
386                         len --;
387                         Sbj[len] = '\0';
388                 }
389                 Encoded = NewStrBufPlain(Sbj, len);
390                 free(Sbj);
391
392                 StrBufTrim(Encoded);
393                 StrBufRFC2047encode(&QPEncoded, Encoded);
394
395                 msg->cm_fields['U'] = SmashStrBuf(&QPEncoded);
396                 FreeStrBuf(&Encoded);
397         }
398         msg->cm_fields['T'] = malloc(64);
399         snprintf(msg->cm_fields['T'], 64, "%ld", ri->pubdate);
400         if (ri->channel_title != NULL) {
401                 if (StrLength(ri->channel_title) > 0) {
402                         msg->cm_fields['O'] = strdup(ChrPtr(ri->channel_title));
403                 }
404         }
405         if (ri->link == NULL) 
406                 ri->link = NewStrBufPlain(HKEY(""));
407
408 #if 0 /* temporarily disable shorter urls. */
409         msg->cm_fields[TMP_SHORTER_URLS] = GetShorterUrls(ri->description);
410 #endif
411
412         msglen += 1024 + StrLength(ri->link) + StrLength(ri->description) ;
413
414         Message = NewStrBufPlain(NULL, StrLength(ri->description));
415
416         StrBufPlain(Message, HKEY(
417                             "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n"
418                             "<html><body>\n"));
419 #if 0 /* disable shorter url for now. */
420         msg->cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message);
421 #endif
422         StrBufAppendBuf(Message, ri->description, 0);
423         StrBufAppendBufPlain(Message, HKEY("<br><br>\n"), 0);
424
425         AppendLink(Message, ri->link, ri->linkTitle, NULL);
426         AppendLink(Message, ri->reLink, ri->reLinkTitle, "Reply to this");
427         StrBufAppendBufPlain(Message, HKEY("</body></html>\n"), 0);
428
429
430
431         networker_save_message *SaveMsg;
432
433         SaveMsg = (networker_save_message *) malloc(sizeof(networker_save_message));
434         memset(SaveMsg, 0, sizeof(networker_save_message));
435         
436         SaveMsg->MsgGUID = guid;
437         SaveMsg->Message = Message;
438         SaveMsg->Msg = msg;
439
440         n = GetCount(Cfg->Messages) + 1;
441         Put(Cfg->Messages, IKEY(n), SaveMsg, FreeNetworkSaveMessage);
442 }
443
444
445
446 /*
447  * Begin a feed parse
448  */
449 int rss_do_fetching(rss_aggregator *Cfg)
450 {
451         rss_item *ri;
452                 
453         time_t now;
454         AsyncIO *IO;
455
456         now = time(NULL);
457
458         if ((Cfg->next_poll != 0) && (now < Cfg->next_poll))
459                 return 0;
460
461         ri = (rss_item*) malloc(sizeof(rss_item));
462         memset(ri, 0, sizeof(rss_item));
463         Cfg->Item = ri;
464         IO = &Cfg->IO;
465         IO->CitContext = CloneContext(&rss_CC);
466         IO->Data = Cfg;
467
468
469         syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(Cfg->Url));
470         ParseURL(&IO->ConnectMe, Cfg->Url, 80);
471         CurlPrepareURL(IO->ConnectMe);
472
473         if (! evcurl_init(IO, 
474 //                        Ctx, 
475                           NULL,
476                           "Citadel RSS Client",
477                           ParseRSSReply, 
478                           RSSAggregatorTerminate))
479         {
480                 syslog(LOG_DEBUG, "Unable to initialize libcurl.\n");
481                 return 0;
482         }
483
484         QueueCurlContext(IO);
485         return 1;
486 }
487
488
489 void DeleteRssCfg(void *vptr)
490 {
491         rss_aggregator *rncptr = (rss_aggregator *)vptr;
492         AsyncIO *IO = &rncptr->IO;
493         EVM_syslog(LOG_DEBUG, "RSS: destroying\n");
494
495         FreeStrBuf(&rncptr->Url);
496         FreeStrBuf(&rncptr->rooms);
497         FreeStrBuf(&rncptr->CData);
498         FreeStrBuf(&rncptr->Key);
499         FreeStrBuf(&rncptr->IO.HttpReq.ReplyData);
500         DeleteHash(&rncptr->OtherQRnumbers);
501         FreeURL(&rncptr->IO.ConnectMe);
502
503         DeleteHashPos (&rncptr->Pos);
504         DeleteHash (&rncptr->Messages);
505         if (rncptr->recp.recp_room != NULL)
506                 free(rncptr->recp.recp_room);
507
508
509         if (rncptr->Item != NULL)
510         {
511                 FreeStrBuf(&rncptr->Item->guid);
512                 FreeStrBuf(&rncptr->Item->title);
513                 FreeStrBuf(&rncptr->Item->link);
514                 FreeStrBuf(&rncptr->Item->linkTitle);
515                 FreeStrBuf(&rncptr->Item->reLink);
516                 FreeStrBuf(&rncptr->Item->reLinkTitle);
517                 FreeStrBuf(&rncptr->Item->description);
518                 FreeStrBuf(&rncptr->Item->channel_title);
519                 FreeStrBuf(&rncptr->Item->author_or_creator);
520                 FreeStrBuf(&rncptr->Item->author_url);
521                 FreeStrBuf(&rncptr->Item->author_email);
522
523                 free(rncptr->Item);
524         }
525         free(rncptr);
526 }
527
528 eNextState RSSAggregatorTerminate(AsyncIO *IO)
529 {
530         rss_aggregator *rncptr = (rss_aggregator *)IO->Data;
531
532         EVM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
533
534
535         UnlinkRSSAggregator(rncptr);
536         return eAbort;
537 }
538
539 /*
540  * Scan a room's netconfig to determine whether it is requesting any RSS feeds
541  */
542 void rssclient_scan_room(struct ctdlroom *qrbuf, void *data)
543 {
544         StrBuf *CfgData=NULL;
545         StrBuf *CfgType;
546         StrBuf *Line;
547         rss_room_counter *Count = NULL;
548         struct stat statbuf;
549         char filename[PATH_MAX];
550         int  fd;
551         int Done;
552         rss_aggregator *rncptr = NULL;
553         rss_aggregator *use_this_rncptr = NULL;
554         void *vptr;
555         const char *CfgPtr, *lPtr;
556         const char *Err;
557
558         pthread_mutex_lock(&RSSQueueMutex);
559         if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
560         {
561                 syslog(LOG_DEBUG, 
562                               "rssclient: [%ld] %s already in progress.\n", 
563                               qrbuf->QRnumber, 
564                               qrbuf->QRname);
565                 pthread_mutex_unlock(&RSSQueueMutex);
566                 return;
567         }
568         pthread_mutex_unlock(&RSSQueueMutex);
569
570         assoc_file_name(filename, sizeof filename, qrbuf, ctdl_netcfg_dir);
571
572         if (server_shutting_down)
573                 return;
574                 
575         /* Only do net processing for rooms that have netconfigs */
576         fd = open(filename, 0);
577         if (fd <= 0) {
578                 //syslog(LOG_DEBUG, "rssclient: %s no config.\n", qrbuf->QRname);
579                 return;
580         }
581
582         if (server_shutting_down)
583                 return;
584
585         if (fstat(fd, &statbuf) == -1) {
586                 syslog(LOG_DEBUG, "ERROR: could not stat configfile '%s' - %s\n",
587                        filename, strerror(errno));
588                 return;
589         }
590
591         if (server_shutting_down)
592                 return;
593
594         CfgData = NewStrBufPlain(NULL, statbuf.st_size + 1);
595
596         if (StrBufReadBLOB(CfgData, &fd, 1, statbuf.st_size, &Err) < 0) {
597                 close(fd);
598                 FreeStrBuf(&CfgData);
599                 syslog(LOG_DEBUG, "ERROR: reading config '%s' - %s<br>\n",
600                         filename, strerror(errno));
601                 return;
602         }
603         close(fd);
604         if (server_shutting_down)
605                 return;
606         
607         CfgPtr = NULL;
608         CfgType = NewStrBuf();
609         Line = NewStrBufPlain(NULL, StrLength(CfgData));
610         Done = 0;
611         while (!Done)
612         {
613             Done = StrBufSipLine(Line, CfgData, &CfgPtr) == 0;
614             if (StrLength(Line) > 0)
615             {
616                 lPtr = NULL;
617                 StrBufExtract_NextToken(CfgType, Line, &lPtr, '|');
618                 if (!strcasecmp("rssclient", ChrPtr(CfgType)))
619                 {
620                     if (Count == NULL)
621                     {
622                         Count = malloc(sizeof(rss_room_counter));
623                         Count->count = 0;
624                     }
625                     Count->count ++;
626                     rncptr = (rss_aggregator *) malloc(sizeof(rss_aggregator));
627                     memset (rncptr, 0, sizeof(rss_aggregator));
628                     rncptr->roomlist_parts = 1;
629                     rncptr->Url = NewStrBuf();
630                     StrBufExtract_NextToken(rncptr->Url, Line, &lPtr, '|');
631
632                     pthread_mutex_lock(&RSSQueueMutex);
633                     GetHash(RSSFetchUrls, SKEY(rncptr->Url), &vptr);
634                     use_this_rncptr = (rss_aggregator *)vptr;
635                     if (use_this_rncptr != NULL)
636                     {
637                             long *QRnumber;
638                             StrBufAppendBufPlain(use_this_rncptr->rooms, 
639                                                  qrbuf->QRname, 
640                                                  -1, 0);
641                             if (use_this_rncptr->roomlist_parts == 1)
642                             {
643                                     use_this_rncptr->OtherQRnumbers = NewHash(1, lFlathash);
644                             }
645                             QRnumber = (long*)malloc(sizeof(long));
646                             *QRnumber = qrbuf->QRnumber;
647                             Put(use_this_rncptr->OtherQRnumbers, LKEY(qrbuf->QRnumber), QRnumber, NULL);
648                             use_this_rncptr->roomlist_parts++;
649
650                             pthread_mutex_unlock(&RSSQueueMutex);
651
652                             FreeStrBuf(&rncptr->Url);
653                             free(rncptr);
654                             rncptr = NULL;
655                             continue;
656                     }
657                     pthread_mutex_unlock(&RSSQueueMutex);
658
659                     rncptr->ItemType = RSS_UNSET;
660                                 
661                     rncptr->rooms = NewStrBufPlain(qrbuf->QRname, -1);
662
663                     pthread_mutex_lock(&RSSQueueMutex);
664                     Put(RSSFetchUrls, SKEY(rncptr->Url), rncptr, DeleteRssCfg);
665                     pthread_mutex_unlock(&RSSQueueMutex);
666                 }
667             }
668         }
669         if (Count != NULL)
670         {
671                 Count->QRnumber = qrbuf->QRnumber;
672                 pthread_mutex_lock(&RSSQueueMutex);
673                 syslog(LOG_DEBUG, "rssclient: [%ld] %s now starting.\n", 
674                               qrbuf->QRnumber, qrbuf->QRname);
675                 Put(RSSQueueRooms, LKEY(qrbuf->QRnumber), Count, NULL);
676                 pthread_mutex_unlock(&RSSQueueMutex);
677         }
678         FreeStrBuf(&CfgData);
679         FreeStrBuf(&CfgType);
680         FreeStrBuf(&Line);
681 }
682
683 /*
684  * Scan for rooms that have RSS client requests configured
685  */
686 void rssclient_scan(void) {
687         static int doing_rssclient = 0;
688         rss_aggregator *rptr = NULL;
689         void *vrptr = NULL;
690         HashPos  *it;
691         long len;
692         const char *Key;
693
694         /* Run no more than once every 15 minutes. */
695         if ((time(NULL) - last_run) < 900) {
696                 return;
697         }
698
699         /*
700          * This is a simple concurrency check to make sure only one rssclient run
701          * is done at a time.  We could do this with a mutex, but since we
702          * don't really require extremely fine granularity here, we'll do it
703          * with a static variable instead.
704          */
705         if (doing_rssclient) return;
706         doing_rssclient = 1;
707         if ((GetCount(RSSQueueRooms) > 0) || (GetCount(RSSFetchUrls) > 0))
708                 return;
709
710         syslog(LOG_DEBUG, "rssclient started\n");
711         CtdlForEachRoom(rssclient_scan_room, NULL);
712
713         pthread_mutex_lock(&RSSQueueMutex);
714
715         it = GetNewHashPos(RSSFetchUrls, 0);
716         while (!server_shutting_down &&
717                GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) && 
718                (vrptr != NULL)) {
719                 rptr = (rss_aggregator *)vrptr;
720                 if (!rss_do_fetching(rptr))
721                         UnlinkRSSAggregator(rptr);
722         }
723         DeleteHashPos(&it);
724         pthread_mutex_unlock(&RSSQueueMutex);
725
726         syslog(LOG_DEBUG, "rssclient ended\n");
727         doing_rssclient = 0;
728         return;
729 }
730
731 void rss_cleanup(void)
732 {
733         /* citthread_mutex_destroy(&RSSQueueMutex); TODO */
734         DeleteHash(&RSSFetchUrls);
735         DeleteHash(&RSSQueueRooms);
736 }
737
738
739 CTDL_MODULE_INIT(rssclient)
740 {
741         if (threading)
742         {
743                 CtdlFillSystemContext(&rss_CC, "rssclient");
744                 pthread_mutex_init(&RSSQueueMutex, NULL);
745                 RSSQueueRooms = NewHash(1, lFlathash);
746                 RSSFetchUrls = NewHash(1, NULL);
747                 syslog(LOG_INFO, "%s\n", curl_version());
748                 CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER);
749                 CtdlRegisterCleanupHook(rss_cleanup);
750         }
751         return "rssclient";
752 }