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)->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         memcpy(CCC->cs_clientname, RSSStates[State].Key, RSSStates[State].len + 1);
 115 }
 116
 117 void DeleteRoomReference(long QRnumber)
 118 {
 119         HashPos *At;
 120         long HKLen;
 121         const char *HK;
 122         void *vData = NULL;
 123         rss_room_counter *pRoomC;
 124
 125         At = GetNewHashPos(RSSQueueRooms, 0);
 126
 127         if (GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At))
 128         {
 129                 GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData);
 130                 if (vData != NULL)
 131                 {
 132                         pRoomC = (rss_room_counter *) vData;
 133                         pRoomC->count --;
 134                         if (pRoomC->count == 0)
 135                                 DeleteEntryFromHash(RSSQueueRooms, At);
 136                 }
 137         }
 138         DeleteHashPos(&At);
 139 }
 140
 141 void UnlinkRooms(rss_aggregator *RSSAggr)
 142 {
 143         DeleteRoomReference(RSSAggr->QRnumber);
 144         if (RSSAggr->OtherQRnumbers != NULL)
 145         {
 146                 long HKLen;
 147                 const char *HK;
 148                 HashPos *At;
 149                 void *vData;
 150
 151                 At = GetNewHashPos(RSSAggr->OtherQRnumbers, 0);
 152                 while (! server_shutting_down &&
 153                        GetNextHashPos(RSSAggr->OtherQRnumbers,
 154                                       At,
 155                                       &HKLen, &HK,
 156                                       &vData) &&
 157                        (vData != NULL))
 158                 {
 159                         long *lData = (long*) vData;
 160                         DeleteRoomReference(*lData);
 161                 }
 162
 163                 DeleteHashPos(&At);
 164         }
 165 }
 166
 167 void UnlinkRSSAggregator(rss_aggregator *RSSAggr)
 168 {
 169         HashPos *At;
 170
 171         pthread_mutex_lock(&RSSQueueMutex);
 172         UnlinkRooms(RSSAggr);
 173
 174         At = GetNewHashPos(RSSFetchUrls, 0);
 175         if (GetHashPosFromKey(RSSFetchUrls, SKEY(RSSAggr->Url), At))
 176         {
 177                 DeleteEntryFromHash(RSSFetchUrls, At);
 178         }
 179         DeleteHashPos(&At);
 180         last_run = time(NULL);
 181         pthread_mutex_unlock(&RSSQueueMutex);
 182 }
 183
 184 void DeleteRssCfg(void *vptr)
 185 {
 186         rss_aggregator *RSSAggr = (rss_aggregator *)vptr;
 187         AsyncIO *IO = &RSSAggr->IO;
 188
 189         if (IO->CitContext != NULL)
 190                 EVRSSCM_syslog(LOG_DEBUG, "RSS: destroying\n");
 191
 192         FreeStrBuf(&RSSAggr->Url);
 193         FreeStrBuf(&RSSAggr->rooms);
 194         FreeStrBuf(&RSSAggr->CData);
 195         FreeStrBuf(&RSSAggr->Key);
 196         DeleteHash(&RSSAggr->OtherQRnumbers);
 197
 198         DeleteHashPos (&RSSAggr->Pos);
 199         DeleteHash (&RSSAggr->Messages);
 200         if (RSSAggr->recp.recp_room != NULL)
 201                 free(RSSAggr->recp.recp_room);
 202
 203
 204         if (RSSAggr->Item != NULL)
 205         {
 206                 flush_rss_item(RSSAggr->Item);
 207
 208                 free(RSSAggr->Item);
 209         }
 210
 211         FreeAsyncIOContents(&RSSAggr->IO);
 212         memset(RSSAggr, 0, sizeof(rss_aggregator));
 213         free(RSSAggr);
 214 }
 215
 216 eNextState RSSAggregator_Terminate(AsyncIO *IO)
 217 {
 218         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
 219
 220         EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
 221
 222         StopCurlWatchers(IO);
 223         UnlinkRSSAggregator(RSSAggr);
 224         return eAbort;
 225 }
 226
 227 eNextState RSSAggregator_TerminateDB(AsyncIO *IO)
 228 {
 229         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
 230
 231         EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
 232
 233
 234         StopDBWatchers(&RSSAggr->IO);
 235         UnlinkRSSAggregator(RSSAggr);
 236         return eAbort;
 237 }
 238
 239 eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO)
 240 {
 241         const char *pUrl;
 242         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
 243
 244         pUrl = IO->ConnectMe->PlainUrl;
 245         if (pUrl == NULL)
 246                 pUrl = "";
 247
 248         EVRSSC_syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.\n", pUrl);
 249
 250         StopCurlWatchers(IO);
 251         UnlinkRSSAggregator(RSSAggr);
 252         return eAbort;
 253 }
 254
 255 void AppendLink(StrBuf *Message,
 256                 StrBuf *link,
 257                 StrBuf *LinkTitle,
 258                 const char *Title)
 259 {
 260         if (StrLength(link) > 0)
 261         {
 262                 StrBufAppendBufPlain(Message, HKEY("<a href=\""), 0);
 263                 StrBufAppendBuf(Message, link, 0);
 264                 StrBufAppendBufPlain(Message, HKEY("\">"), 0);
 265                 if (StrLength(LinkTitle) > 0)
 266                         StrBufAppendBuf(Message, LinkTitle, 0);
 267                 else if ((Title != NULL) && !IsEmptyStr(Title))
 268                         StrBufAppendBufPlain(Message, Title, -1, 0);
 269                 else
 270                         StrBufAppendBuf(Message, link, 0);
 271                 StrBufAppendBufPlain(Message, HKEY("</a><br>\n"), 0);
 272         }
 273 }
 274
 275
 276 void rss_format_item(networker_save_message *SaveMsg)
 277 {
 278         StrBuf *Message;
 279         int msglen = 0;
 280
 281         if (SaveMsg->author_or_creator != NULL) {
 282
 283                 char *From;
 284                 StrBuf *Encoded = NULL;
 285                 int FromAt;
 286
 287                 From = html_to_ascii(ChrPtr(SaveMsg->author_or_creator),
 288                                      StrLength(SaveMsg->author_or_creator),
 289                                      512, 0);
 290                 StrBufPlain(SaveMsg->author_or_creator, From, -1);
 291                 StrBufTrim(SaveMsg->author_or_creator);
 292                 free(From);
 293
 294                 FromAt = strchr(ChrPtr(SaveMsg->author_or_creator), '@') != NULL;
 295                 if (!FromAt && StrLength (SaveMsg->author_email) > 0)
 296                 {
 297                         StrBufRFC2047encode(&Encoded, SaveMsg->author_or_creator);
 298                         SaveMsg->Msg.cm_fields['A'] = SmashStrBuf(&Encoded);
 299                         SaveMsg->Msg.cm_fields['P'] =
 300                                 SmashStrBuf(&SaveMsg->author_email);
 301                 }
 302                 else
 303                 {
 304                         if (FromAt)
 305                         {
 306                                 SaveMsg->Msg.cm_fields['A'] =
 307                                         SmashStrBuf(&SaveMsg->author_or_creator);
 308                                 SaveMsg->Msg.cm_fields['P'] =
 309                                         strdup(SaveMsg->Msg.cm_fields['A']);
 310                         }
 311                         else
 312                         {
 313                                 StrBufRFC2047encode(&Encoded,
 314                                                     SaveMsg->author_or_creator);
 315                                 SaveMsg->Msg.cm_fields['A'] =
 316                                         SmashStrBuf(&Encoded);
 317                                 SaveMsg->Msg.cm_fields['P'] =
 318                                         strdup("rss@localhost");
 319
 320                         }
 321                 }
 322         }
 323         else {
 324                 SaveMsg->Msg.cm_fields['A'] = strdup("rss");
 325         }
 326
 327         SaveMsg->Msg.cm_fields['N'] = strdup(NODENAME);
 328         if (SaveMsg->title != NULL) {
 329                 long len;
 330                 char *Sbj;
 331                 StrBuf *Encoded, *QPEncoded;
 332
 333                 QPEncoded = NULL;
 334                 StrBufSpaceToBlank(SaveMsg->title);
 335                 len = StrLength(SaveMsg->title);
 336                 Sbj = html_to_ascii(ChrPtr(SaveMsg->title), len, 512, 0);
 337                 len = strlen(Sbj);
 338                 if ((len > 0) && (Sbj[len - 1] == '\n'))
 339                 {
 340                         len --;
 341                         Sbj[len] = '\0';
 342                 }
 343                 Encoded = NewStrBufPlain(Sbj, len);
 344                 free(Sbj);
 345
 346                 StrBufTrim(Encoded);
 347                 StrBufRFC2047encode(&QPEncoded, Encoded);
 348
 349                 SaveMsg->Msg.cm_fields['U'] = SmashStrBuf(&QPEncoded);
 350                 FreeStrBuf(&Encoded);
 351         }
 352         if (SaveMsg->link == NULL)
 353                 SaveMsg->link = NewStrBufPlain(HKEY(""));
 354
 355 #if 0 /* temporarily disable shorter urls. */
 356         SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] =
 357                 GetShorterUrls(SaveMsg->description);
 358 #endif
 359
 360         msglen += 1024 + StrLength(SaveMsg->link) + StrLength(SaveMsg->description) ;
 361
 362         Message = NewStrBufPlain(NULL, msglen);
 363
 364         StrBufPlain(Message, HKEY(
 365                             "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n"
 366                             "<html><body>\n"));
 367 #if 0 /* disable shorter url for now. */
 368         SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message);
 369 #endif
 370         StrBufAppendBuf(Message, SaveMsg->description, 0);
 371         StrBufAppendBufPlain(Message, HKEY("<br><br>\n"), 0);
 372
 373         AppendLink(Message, SaveMsg->link, SaveMsg->linkTitle, NULL);
 374         AppendLink(Message, SaveMsg->reLink, SaveMsg->reLinkTitle, "Reply to this");
 375         StrBufAppendBufPlain(Message, HKEY("</body></html>\n"), 0);
 376
 377
 378         SaveMsg->Message = Message;
 379 }
 380
 381 eNextState RSSSaveMessage(AsyncIO *IO)
 382 {
 383         long len;
 384         const char *Key;
 385         rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data;
 386
 387         rss_format_item(RSSAggr->ThisMsg);
 388
 389         RSSAggr->ThisMsg->Msg.cm_fields['M'] =
 390                 SmashStrBuf(&RSSAggr->ThisMsg->Message);
 391
 392         CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0);
 393
 394         /* write the uidl to the use table so we don't store this item again */
 395
 396         CheckIfAlreadySeen("RSS Item Insert", RSSAggr->ThisMsg->MsgGUID, IO->Now, 0, eWrite, IO->ID, CCID);
 397
 398         if (GetNextHashPos(RSSAggr->Messages,
 399                            RSSAggr->Pos,
 400                            &len, &Key,
 401                            (void**) &RSSAggr->ThisMsg))
 402                 return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
 403         else
 404                 return eAbort;
 405 }
 406
 407 eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
 408 {
 409         const char *Key;
 410         long len;
 411         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
 412
 413         /* Find out if we've already seen this item */
 414 // todo: expiry?
 415 #ifndef DEBUG_RSS
 416         SetRSSState(IO, eRSSUT);
 417         if (CheckIfAlreadySeen("RSS Item Seen",
 418                                Ctx->ThisMsg->MsgGUID,
 419                                IO->Now,
 420                                IO->Now - USETABLE_ANTIEXPIRE,
 421                                eCheckUpdate,
 422                                IO->ID, CCID)
 423             != 0)
 424         {
 425                 /* Item has already been seen */
 426                 EVRSSC_syslog(LOG_DEBUG,
 427                           "%s has already been seen\n",
 428                           ChrPtr(Ctx->ThisMsg->MsgGUID));
 429                 SetRSSState(IO, eRSSParsing);
 430
 431                 if (GetNextHashPos(Ctx->Messages,
 432                                    Ctx->Pos,
 433                                    &len, &Key,
 434                                    (void**) &Ctx->ThisMsg))
 435                         return NextDBOperation(
 436                                 IO,
 437                                 RSS_FetchNetworkUsetableEntry);
 438                 else
 439                         return eAbort;
 440         }
 441         else
 442 #endif
 443         {
 444                 SetRSSState(IO, eRSSParsing);
 445
 446                 NextDBOperation(IO, RSSSaveMessage);
 447                 return eSendMore;
 448         }
 449         return eSendMore;
 450 }
 451
 452 eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
 453 {
 454         u_char rawdigest[MD5_DIGEST_LEN];
 455         struct MD5Context md5context;
 456         StrBuf *guid;
 457         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
 458
 459         if (IO->HttpReq.httpcode != 200)
 460         {
 461                 StrBuf *ErrMsg;
 462                 long lens[2];
 463                 const char *strs[2];
 464
 465                 SetRSSState(IO, eRSSFailure);
 466                 ErrMsg = NewStrBuf();
 467                 EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n",
 468                               IO->HttpReq.httpcode);
 469
 470                 strs[0] = ChrPtr(Ctx->Url);
 471                 lens[0] = StrLength(Ctx->Url);
 472
 473                 strs[1] = ChrPtr(Ctx->rooms);
 474                 lens[1] = StrLength(Ctx->rooms);
 475                 StrBufPrintf(ErrMsg,
 476                              "Error while RSS-Aggregation Run of %s\n"
 477                              " need a 200, got a %ld !\n"
 478                              " Response text was: \n"
 479                              " \n %s\n",
 480                              ChrPtr(Ctx->Url),
 481                              IO->HttpReq.httpcode,
 482                              ChrPtr(IO->HttpReq.ReplyData));
 483                 CtdlAideFPMessage(
 484                         ChrPtr(ErrMsg),
 485                         "RSS Aggregation run failure",
 486                         2, strs, (long*) &lens,
 487                         IO->Now,
 488                         IO->ID, CCID);
 489
 490                 FreeStrBuf(&ErrMsg);
 491                 EVRSSC_syslog(LOG_DEBUG,
 492                               "RSS feed returned an invalid http status code. <%s><HTTP %ld>\n",
 493                               ChrPtr(Ctx->Url),
 494                               IO->HttpReq.httpcode);
 495                 return eAbort;
 496         }
 497         SetRSSState(IO, eRSSUT);
 498
 499         MD5Init(&md5context);
 500
 501         MD5Update(&md5context,
 502                   (const unsigned char*)SKEY(IO->HttpReq.ReplyData));
 503
 504         MD5Update(&md5context,
 505                   (const unsigned char*)SKEY(Ctx->Url));
 506
 507         MD5Final(rawdigest, &md5context);
 508         guid = NewStrBufPlain(NULL,
 509                               MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/);
 510         StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN);
 511         StrBufAppendBufPlain(guid, HKEY("_rssFM"), 0);
 512         if (StrLength(guid) > 40)
 513                 StrBufCutAt(guid, 40, NULL);
 514         /* Find out if we've already seen this item */
 515
 516 #ifndef DEBUG_RSS
 517
 518         if (CheckIfAlreadySeen("RSS Whole",
 519                                guid,
 520                                IO->Now,
 521                                IO->Now - USETABLE_ANTIEXPIRE,
 522                                eCheckUpdate,
 523                                IO->ID, CCID)
 524             != 0)
 525         {
 526                 FreeStrBuf(&guid);
 527
 528                 EVRSSC_syslog(LOG_DEBUG, "RSS feed already seen. <%s>\n", ChrPtr(Ctx->Url));
 529                 return eAbort;
 530         }
 531         FreeStrBuf(&guid);
 532 #endif
 533         SetRSSState(IO, eRSSParsing);
 534         return RSSAggregator_ParseReply(IO);
 535 }
 536
 537 eNextState RSSAggregator_FinishHttp(AsyncIO *IO)
 538 {
 539         return QueueDBOperation(IO, RSSAggregator_AnalyseReply);
 540 }
 541
 542 /*
 543  * Begin a feed parse
 544  */
 545 int rss_do_fetching(rss_aggregator *RSSAggr)
 546 {
 547         AsyncIO         *IO = &RSSAggr->IO;
 548         rss_item *ri;
 549         time_t now;
 550
 551         now = time(NULL);
 552
 553         if ((RSSAggr->next_poll != 0) && (now < RSSAggr->next_poll))
 554                 return 0;
 555
 556         ri = (rss_item*) malloc(sizeof(rss_item));
 557         memset(ri, 0, sizeof(rss_item));
 558         RSSAggr->Item = ri;
 559
 560         if (! InitcURLIOStruct(&RSSAggr->IO,
 561                                RSSAggr,
 562                                "Citadel RSS Client",
 563                                RSSAggregator_FinishHttp,
 564                                RSSAggregator_Terminate,
 565                                RSSAggregator_TerminateDB,
 566                                RSSAggregator_ShutdownAbort))
 567         {
 568                 EVRSSCM_syslog(LOG_ALERT, "Unable to initialize libcurl.\n");
 569                 return 0;
 570         }
 571         SetRSSState(IO, eRSSCreated);
 572
 573         safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host,
 574                     ChrPtr(RSSAggr->Url),
 575                     sizeof(((CitContext*)RSSAggr->IO.CitContext)->cs_host));
 576
 577         EVRSSC_syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(RSSAggr->Url));
 578         ParseURL(&RSSAggr->IO.ConnectMe, RSSAggr->Url, 80);
 579         CurlPrepareURL(RSSAggr->IO.ConnectMe);
 580
 581         SetRSSState(IO, eRSSFetching);
 582         QueueCurlContext(&RSSAggr->IO);
 583         return 1;
 584 }
 585
 586 /*
 587  * Scan a room's netconfig to determine whether it is requesting any RSS feeds
 588  */
 589 void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
 590 {
 591         const RoomNetCfgLine *pLine;
 592         rss_aggregator *RSSAggr = NULL;
 593         rss_aggregator *use_this_RSSAggr = NULL;
 594         void *vptr;
 595
 596         pthread_mutex_lock(&RSSQueueMutex);
 597         if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
 598         {
 599                 EVRSSQ_syslog(LOG_DEBUG,
 600                               "rssclient: [%ld] %s already in progress.\n",
 601                               qrbuf->QRnumber,
 602                               qrbuf->QRname);
 603                 pthread_mutex_unlock(&RSSQueueMutex);
 604                 return;
 605         }
 606         pthread_mutex_unlock(&RSSQueueMutex);
 607
 608         if (server_shutting_down) return;
 609
 610         pLine = OneRNCFG->NetConfigs[rssclient];
 611
 612         while (pLine != NULL)
 613         {
 614                 const char *lPtr = NULL;
 615
 616                 RSSAggr = (rss_aggregator *) malloc(
 617                         sizeof(rss_aggregator));
 618
 619                 memset (RSSAggr, 0, sizeof(rss_aggregator));
 620                 RSSAggr->QRnumber = qrbuf->QRnumber;
 621                 RSSAggr->roomlist_parts = 1;
 622                 RSSAggr->Url = NewStrBufPlain(NULL, StrLength(pLine->Value[0]));
 623                 StrBufExtract_NextToken(RSSAggr->Url,
 624                                         pLine->Value[0],
 625                                         &lPtr,
 626                                         '|');
 627
 628                 pthread_mutex_lock(&RSSQueueMutex);
 629                 GetHash(RSSFetchUrls,
 630                         SKEY(RSSAggr->Url),
 631                         &vptr);
 632
 633                 use_this_RSSAggr = (rss_aggregator *)vptr;
 634                 if (use_this_RSSAggr != NULL)
 635                 {
 636                         long *QRnumber;
 637                         StrBufAppendBufPlain(
 638                                 use_this_RSSAggr->rooms,
 639                                 qrbuf->QRname,
 640                                 -1, 0);
 641                         if (use_this_RSSAggr->roomlist_parts==1)
 642                         {
 643                                 use_this_RSSAggr->OtherQRnumbers
 644                                         = NewHash(1, lFlathash);
 645                         }
 646                         QRnumber = (long*)malloc(sizeof(long));
 647                         *QRnumber = qrbuf->QRnumber;
 648                         Put(use_this_RSSAggr->OtherQRnumbers,
 649                             LKEY(qrbuf->QRnumber),
 650                             QRnumber,
 651                             NULL);
 652                         use_this_RSSAggr->roomlist_parts++;
 653
 654                         pthread_mutex_unlock(&RSSQueueMutex);
 655
 656                         FreeStrBuf(&RSSAggr->Url);
 657                         free(RSSAggr);
 658                         RSSAggr = NULL;
 659                         pLine = pLine->next;
 660                         continue;
 661                 }
 662                 pthread_mutex_unlock(&RSSQueueMutex);
 663
 664                 RSSAggr->ItemType = RSS_UNSET;
 665
 666                 RSSAggr->rooms = NewStrBufPlain(
 667                         qrbuf->QRname, -1);
 668
 669                 pthread_mutex_lock(&RSSQueueMutex);
 670
 671                 Put(RSSFetchUrls,
 672                     SKEY(RSSAggr->Url),
 673                     RSSAggr,
 674                     DeleteRssCfg);
 675
 676                 pthread_mutex_unlock(&RSSQueueMutex);
 677                 pLine = pLine->next;
 678         }
 679 }
 680
 681 /*
 682  * Scan for rooms that have RSS client requests configured
 683  */
 684 void rssclient_scan(void) {
 685         int RSSRoomCount, RSSCount;
 686         rss_aggregator *rptr = NULL;
 687         void *vrptr = NULL;
 688         HashPos *it;
 689         long len;
 690         const char *Key;
 691         time_t now = time(NULL);
 692
 693         /* Run no more than once every 15 minutes. */
 694         if ((now - last_run) < 900) {
 695                 EVRSSQ_syslog(LOG_DEBUG,
 696                               "Client: polling interval not yet reached; last run was %ldm%lds ago",
 697                               ((now - last_run) / 60),
 698                               ((now - last_run) % 60)
 699                 );
 700                 return;
 701         }
 702
 703         /*
 704          * This is a simple concurrency check to make sure only one rssclient
 705          * run is done at a time.
 706          */
 707         pthread_mutex_lock(&RSSQueueMutex);
 708         RSSCount = GetCount(RSSFetchUrls);
 709         RSSRoomCount = GetCount(RSSQueueRooms);
 710         pthread_mutex_unlock(&RSSQueueMutex);
 711
 712         if ((RSSRoomCount > 0) || (RSSCount > 0)) {
 713                 EVRSSQ_syslog(LOG_DEBUG,
 714                               "rssclient: concurrency check failed; %d rooms and %d url's are queued",
 715                               RSSRoomCount, RSSCount
 716                         );
 717                 return;
 718         }
 719
 720         become_session(&rss_CC);
 721         EVRSSQM_syslog(LOG_DEBUG, "rssclient started");
 722         CtdlForEachNetCfgRoom(rssclient_scan_room, NULL, rssclient);
 723
 724         if (GetCount(RSSFetchUrls) > 0)
 725         {
 726                 pthread_mutex_lock(&RSSQueueMutex);
 727                 EVRSSQ_syslog(LOG_DEBUG,
 728                                "rssclient starting %d Clients",
 729                                GetCount(RSSFetchUrls));
 730
 731                 it = GetNewHashPos(RSSFetchUrls, 0);
 732                 while (!server_shutting_down &&
 733                        GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) &&
 734                        (vrptr != NULL)) {
 735                         rptr = (rss_aggregator *)vrptr;
 736                         if (!rss_do_fetching(rptr))
 737                                 UnlinkRSSAggregator(rptr);
 738                 }
 739                 DeleteHashPos(&it);
 740                 pthread_mutex_unlock(&RSSQueueMutex);
 741         }
 742         else
 743                 EVRSSQM_syslog(LOG_DEBUG, "Nothing to do.");
 744
 745         EVRSSQM_syslog(LOG_DEBUG, "rssclient ended\n");
 746         return;
 747 }
 748
 749 void rss_cleanup(void)
 750 {
 751         /* citthread_mutex_destroy(&RSSQueueMutex); TODO */
 752         DeleteHash(&RSSFetchUrls);
 753         DeleteHash(&RSSQueueRooms);
 754 }
 755
 756 void LogDebugEnableRSSClient(const int n)
 757 {
 758         RSSClientDebugEnabled = n;
 759 }
 760
 761 CTDL_MODULE_INIT(rssclient)
 762 {
 763         if (!threading)
 764         {
 765                 CtdlREGISTERRoomCfgType(rssclient, ParseGeneric, 0, 1, SerializeGeneric, DeleteGenericCfgLine); /// todo: implement rss specific parser
 766                 pthread_mutex_init(&RSSQueueMutex, NULL);
 767                 RSSQueueRooms = NewHash(1, lFlathash);
 768                 RSSFetchUrls = NewHash(1, NULL);
 769                 syslog(LOG_INFO, "%s\n", curl_version());
 770                 CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER, PRIO_AGGR + 300);
 771                 CtdlRegisterEVCleanupHook(rss_cleanup);
 772                 CtdlRegisterDebugFlagHook(HKEY("rssclient"), LogDebugEnableRSSClient, &RSSClientDebugEnabled);
 773         }
 774         else
 775         {
 776                 CtdlFillSystemContext(&rss_CC, "rssclient");
 777         }
 778         return "rssclient";
 779 }