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