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