src/lib/PollManager.py

   1 """ PollManager.py
   2
   3 Module for polling the feeds.
   4 """
   5 __copyright__ = "Copyright (c) 2002-4 Juri Pakaste <juri@iki.fi>"
   6 __license__ = """ GNU General Public License
   7
   8 Straw is free software; you can redistribute it and/or modify it under the
   9 terms of the GNU General Public License as published by the Free Software
  10 Foundation; either version 2 of the License, or (at your option) any later
  11 version.
  12
  13 Straw is distributed in the hope that it will be useful, but WITHOUT ANY
  14 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
  15 A PARTICULAR PURPOSE. See the GNU General Public License for more details.
  16
  17 You should have received a copy of the GNU General Public License along with
  18 this program; if not, write to the Free Software Foundation, Inc., 59 Temple
  19 Place - Suite 330, Boston, MA 02111-1307, USA. """
  20
  21 import Config
  22 import feeds
  23 import Event
  24 from MainloopManager import MainloopManager
  25 import MessageManager
  26 import NetworkConstants
  27 import URLFetch
  28 import FeedCategoryList
  29 import SummaryParser
  30 import utils
  31 import time
  32 import socket
  33 import error
  34
  35 class PollStopper(object, Event.SignalEmitter):
  36     def __init__(self, stopper, object):
  37         Event.SignalEmitter.__init__(self)
  38         self.initialize_slots(Event.PollingStoppedSignal)
  39         self._stopper = stopper
  40         self._object = object
  41
  42     stopper = property(lambda self: self._stopper)
  43     object = property(lambda self: self._object)
  44
  45     def stop(self):
  46         if self._stopper is not None:
  47             self._stopper()
  48             self._stopper = None
  49             self.emit_signal(Event.PollingStoppedSignal(self))
  50
  51 class PollManager:
  52     def __init__(self):
  53         self._config = Config.get_instance()
  54         self._mmgr = MessageManager.get_instance()
  55         self._feedlist = feeds.get_instance()
  56
  57         self._config.signal_connect(Event.PollFrequencyChangedSignal, self.poll_changed)
  58         self._config.signal_connect(Event.OfflineModeChangedSignal, self.offline_changed)
  59
  60         self._feedlist.connect('changed', self.feeds_changed)
  61
  62         self._is_offline = self._config.offline
  63         self._poll_frequency = self._config.poll_frequency
  64         self._last_poll = self._config.last_poll
  65         self._feeds = self._feedlist.flatten_list()
  66         self._pollers = {}
  67
  68     def poll_changed(self, signal):
  69         """ called when global poll frequency has changed"""
  70         self._poll_frequency = signal.value
  71
  72     def offline_changed(self, signal):
  73         self._is_offline = self._config.offline
  74
  75     def feeds_changed(self, *args):
  76         self._feedlist = feeds.get_instance()
  77         self._feeds = self._feedlist.flatten_list()
  78
  79     def start_polling_loop(self):
  80         mlmgr = MainloopManager.get_instance()
  81         mlmgr.set_repeating_timer(NetworkConstants.POLL_INTERVAL, self.maybe_poll)
  82
  83     ##########################
  84     # Why both pollcontext and self._pollers?
  85     # self._pollers is there to make sure that there are no simultaneous
  86     # pollers on the same object. So the user doesn't accidentally start
  87     # a poller for an object being polled.
  88     # pollcontext is there so that we can avoid starting pollers many
  89     # times - even though they wouldn't overlap - on the same object within
  90     # one polling run.
  91
  92     # because maybe_poll is used as an idle handler, it must always return
  93     # True or else it's removed from the handler list
  94     def maybe_poll(self):
  95         self._maybe_poll()
  96
  97     def _maybe_poll(self):
  98         now = int(time.time())
  99         if self._is_offline:
 100             return True
 101         try:
 102             context = {}
 103             config_pf = self._config.poll_frequency
 104             feed_use_default = feeds.Feed.DEFAULT
 105             self.poll([feed for feed, timediff, fpf in
 106                        [(feed, now - feed.last_poll, feed.poll_frequency)
 107                         for feed in self._feeds]
 108                        if ((timediff > fpf > 0) or
 109                            (fpf == feed_use_default and
 110                             timediff > config_pf > 0))], context)
 111
 112
 113             self.poll_categories([cat for cat, sub, timediff in
 114             [(cat, cat.subscription, now - cat.subscription.last_poll)
 115              for cat in FeedCategoryList.get_instance().all_categories
 116              if cat.subscription is not None]
 117             if ((timediff > sub.frequency > 0) or
 118                 (sub.frequency == sub.REFRESH_DEFAULT and
 119                  timediff > config_pf > 0))], context, False)
 120         except:
 121             error.log_exc("Caught an exception while polling")
 122         return True
 123
 124     def poll(self, obj, pollcontext = None):
 125         """ obj must be a list of Feed objects"""
 126         if pollcontext is None:
 127             pollcontext = {}
 128         stoppers = []
 129         for f in obj:
 130             sf = None
 131             if not self._pollers.has_key(f) and not pollcontext.has_key(f):
 132                 self._pollers[f] = True
 133                 pollcontext[f] = True
 134                 sf = FeedPoller(f).poll()
 135             stoppers.append(sf)
 136         return stoppers
 137
 138     def poll_categories(self, cats, pollcontext = None, feeds = True):
 139         """Polls categories and, if feeds is True, the feeds
 140         associated with them.  Returns a list of list containing
 141         PollStopper objects.  poll_categories([c1, c2, c3]) -> [[sc1,
 142         sc1f1, sc1f2 ...], [sc2, sc2f1, sc2f2 ...], ...] where cN is a
 143         category, scN is a PollStopper for cN, scNfM is a stopper for
 144         a feed in cN.  If no polling was started for that cN or cNfM,
 145         in its place is None.  """
 146         stoppers = []
 147         if pollcontext is None:
 148             pollcontext = {}
 149         for c in cats:
 150             category_stoppers = []
 151             sc = None
 152             if c.subscription is not None and not self._pollers.has_key(c) and not pollcontext.has_key(c):
 153                 self._pollers[c] = True
 154                 pollcontext[c] = True
 155                 sc = CategoryPoller(c, pollcontext).poll()
 156             category_stoppers.append(sc)
 157             if feeds:
 158                 for f in c.feeds:
 159                     sf = None
 160                     if not self._pollers.has_key(f) and not pollcontext.has_key(f):
 161                         self._pollers[f] = True
 162                         pollcontext[f] = True
 163                         sf = FeedPoller(f).poll(), f
 164                     category_stoppers.append(sf)
 165             stoppers.append(category_stoppers)
 166         return stoppers
 167
 168     def poll_done(self, obj):
 169         if self._pollers.has_key(obj):
 170             del self._pollers[obj]
 171
 172 class FeedPoller:
 173     def __init__(self, feed):
 174         self._feed = feed
 175         self._mmgr = MessageManager.get_instance()
 176
 177     def poll(self):
 178         MainloopManager.get_instance().call_pending()
 179
 180         url, user, pw = self._feed.access_info
 181         parsed = None
 182         config = Config.get_instance()
 183         self._feed.last_poll = int (time.time())
 184         ps = None
 185         try:
 186             try:
 187                 stopper = URLFetch.get_instance().request(
 188                     url, self, user, pw, priority=NetworkConstants.PRIORITY_RSS)
 189                 ps = PollStopper(stopper, self._feed)
 190             except Exception, e:
 191                 error.log_exc("Caught an exception while polling")
 192                 self.http_failed(e)
 193             else:
 194                 self._feed.router.start_polling()
 195                 self._feed.router.stopper = ps
 196         finally:
 197             return ps
 198
 199     def http_results(self, status, data):
 200         MainloopManager.get_instance().call_pending()
 201         try:
 202             try:
 203                 err = ""
 204                 read_data = True
 205                 #if status is None:
 206                 #    err = _("No data")
 207                 #elif status[1] == 304:
 208                 #    self._feed.router.route_no_data()
 209                 #    read_data = False
 210                 #elif status[1] == 410 or status[1] == 404:
 211                 #    err = _("Unable to find the feed (%s: %s)") % (
 212                 #        status[1], status[2].strip())
 213                 #elif status[1] == 401:
 214                 #    err = _("Invalid username and password.")
 215                 #elif status[1] > 299:
 216                 #    err = _("Updating feed resulted in abnormal status '%s' (code %d)") % (
 217                 #        status[2].strip(), status[1])
 218
 219                 (x, code, message) = status
 220                 if code == 304:
 221                     self._feed.router.route_no_data()
 222                     read_data = False
 223                 elif code > 299:
 224                     err = "%s" % message
 225
 226                 if err:
 227                     self._feed.router.set_error(err)
 228                 elif read_data:
 229                     try:
 230                         parsed = SummaryParser.parse(data, self._feed)
 231                     except Exception, e:
 232                         error.log_exc("exception in summaryparser")
 233                         self._feed.router.set_error(
 234                             _("An error occurred while processing feed: %s") %
 235                             str(e))
 236                     else:
 237                         self._feed.router.route_all(parsed)
 238                 self._mmgr.post_message(_("Updating %s done.") % self._feed.title)
 239             except Exception, ex:
 240                 error.log_exc("error while parsing results")
 241                 self._feed.router.set_error(str(ex))
 242         finally:
 243             get_instance().poll_done(self._feed)
 244
 245     def http_failed(self, exception):
 246         try:
 247             if isinstance(exception, socket.error):
 248                 self._feed.router.route_no_data()
 249             else:
 250                 self._feed.router.set_error(str(exception))
 251             self._mmgr.post_message(_("Updating %s failed") % self._feed.title)
 252             self._feed.poll_done()
 253         finally:
 254             get_instance().poll_done(self._feed)
 255
 256     def http_permanent_redirect(self, location):
 257         # XXX
 258         (oldloc, u, p) = self._feed.access_info
 259         self._feed.access_info = (location, u, p)
 260
 261     def operation_stopped(self):
 262         self._feed.router.route_no_data()
 263         self._feed.poll_done()
 264         get_instance().poll_done(self._feed)
 265
 266 class CategoryPoller:
 267     def __init__(self, category, pollcontext):
 268         self._category = category
 269         self._mmgr = MessageManager.get_instance()
 270         self._pollcontext = pollcontext
 271
 272     def poll(self):
 273         sub = self._category.subscription
 274         if sub is None or sub.location is None or len(sub.location) == 0:
 275             return None
 276         sub.last_poll = int(time.time())
 277         ps = None
 278         try:
 279             try:
 280                 stopper = URLFetch.get_instance().request(
 281                     sub.location, self, sub.username, sub.password, priority=NetworkConstants.PRIORITY_RSS)
 282                 ps = PollStopper(stopper, self._category)
 283             except Exception, e:
 284                 error.log_exc("Caught an exception while polling category")
 285         finally:
 286             return ps
 287
 288     def http_results(self, status,  data):
 289         sub = self._category.subscription
 290         try:
 291             try:
 292                 err = None
 293                 (x, code, message) = status
 294                 if code == 304:
 295                     self._feed.router.route_no_data()
 296                     read_data = False
 297                 elif code > 299:
 298                     err = "%s" % message
 299
 300                 if err is not None:
 301                     self._category.subscription.error = err
 302                 else:
 303                     sub.parse(data)
 304                     old_feeds = self._category.feeds
 305                     self._category.read_contents_from_subscription()
 306                     common, inold, innew = utils.listdiff(
 307                         old_feeds, self._category.feeds)
 308                     if len(innew) > 0:
 309                         get_instance().poll(innew, self._pollcontext)
 310             except Exception, e:
 311                 self._category.subscription.error = str(e)
 312         finally:
 313             get_instance().poll_done(self._category)
 314
 315     def http_failed(self, exception):
 316         self._category.subscription.error = str(exception)
 317         get_instance().poll_done(self._category)
 318
 319     def http_permanent_redirect(self, location):
 320         error.logparam(locals(), "location")
 321
 322     def operation_stopped(self):
 323         get_instance().poll_done(self._category)
 324
 325 pollmanager_instance = None
 326
 327 def get_instance():
 328     global pollmanager_instance
 329     if pollmanager_instance is None:
 330         pollmanager_instance = PollManager()
 331     return pollmanager_instance