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