straw/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 gobject
  22 import Config
  23 from MainloopManager import MainloopManager
  24 import NetworkConstants
  25 import URLFetch
  26 import SummaryParser
  27 import time
  28 import socket
  29 import error
  30
  31 from MessageManager import post_status_message
  32 from straw import helpers
  33
  34 class PollStopper(gobject.GObject):
  35
  36     __gsignals__ = {
  37         'stopped' : (gobject.SIGNAL_RUN_LAST, gobject.TYPE_NONE, ())
  38         }
  39
  40     def __init__(self, stopper, object):
  41         gobject.GObject.__init__(self)
  42         self._stopper = stopper
  43         self._object = object
  44
  45     stopper = property(lambda self: self._stopper)
  46     object = property(lambda self: self._object)
  47
  48     def stop(self):
  49         if self._stopper is not None:
  50             self._stopper()
  51             self._stopper = None
  52             self.emit_signal('stopped')
  53
  54 class PollManager:
  55     def __init__(self):
  56         config = Config.get_instance()
  57         self._feedlist = feeds.feedlist
  58
  59         config.connect('refresh-changed', self.poll_changed)
  60         config.connect('offline-mode-changed', self.offline_changed)
  61
  62         self._feedlist.connect('changed', self.feeds_changed)
  63
  64         self._is_offline = config.offline
  65         self._poll_frequency = config.poll_frequency
  66         self._last_poll = config.last_poll
  67         self._feeds = self._feedlist.flatten_list()
  68         self._pollers = {}
  69
  70     def poll_changed(self, config):
  71         """ called when global poll frequency has changed"""
  72         self._poll_frequency = config.poll_frequency
  73
  74     def offline_changed(self, config):
  75         self._is_offline = config.offline
  76
  77     def feeds_changed(self, *args):
  78         self._feedlist = feeds.feedlist
  79         self._feeds = self._feedlist.flatten_list()
  80
  81     def start_polling_loop(self):
  82         mlmgr = MainloopManager.get_instance()
  83         mlmgr.set_repeating_timer(NetworkConstants.POLL_INTERVAL, self.maybe_poll)
  84
  85     ##########################
  86     # Why both pollcontext and self._pollers?
  87     # self._pollers is there to make sure that there are no simultaneous
  88     # pollers on the same object. So the user doesn't accidentally start
  89     # a poller for an object being polled.
  90     # pollcontext is there so that we can avoid starting pollers many
  91     # times - even though they wouldn't overlap - on the same object within
  92     # one polling run.
  93
  94     # because maybe_poll is used as an idle handler, it must always return
  95     # True or else it's removed from the handler list
  96     def maybe_poll(self):
  97         self._maybe_poll()
  98
  99     def _maybe_poll(self):
 100         now = int(time.time())
 101         if self._is_offline:
 102             return True
 103         try:
 104             context = {}
 105             config_pf = self._poll_frequency
 106             feed_use_default = feeds.Feed.DEFAULT
 107             self.poll([feed for feed, timediff, fpf in
 108                        [(feed, now - feed.last_poll, feed.poll_frequency)
 109                         for feed in self._feeds]
 110                        if ((timediff > fpf > 0) or
 111                            (fpf == feed_use_default and
 112                             timediff > config_pf > 0))], context)
 113
 114
 115             self.poll_categories([cat for cat, sub, timediff in
 116             [(cat, cat.subscription, now - cat.subscription.last_poll)
 117              for cat in feeds.category_list.all_categories
 118              if cat.subscription is not None]
 119             if ((timediff > sub.frequency > 0) or
 120                 (sub.frequency == sub.REFRESH_DEFAULT and
 121                  timediff > config_pf > 0))], context, False)
 122         except:
 123             error.log_exc("Caught an exception while polling")
 124         return True
 125
 126     def poll(self, obj, pollcontext = None):
 127         """ obj must be a list of Feed objects"""
 128         if pollcontext is None:
 129             pollcontext = {}
 130         stoppers = []
 131         for f in obj:
 132             sf = None
 133             if not self._pollers.has_key(f) and not pollcontext.has_key(f):
 134                 self._pollers[f] = True
 135                 pollcontext[f] = True
 136                 sf = FeedPoller(f).poll()
 137             stoppers.append(sf)
 138         return stoppers
 139
 140     def poll_categories(self, cats, pollcontext = None, feeds = True):
 141         """Polls categories and, if feeds is True, the feeds
 142         associated with them.  Returns a list of list containing
 143         PollStopper objects.  poll_categories([c1, c2, c3]) -> [[sc1,
 144         sc1f1, sc1f2 ...], [sc2, sc2f1, sc2f2 ...], ...] where cN is a
 145         category, scN is a PollStopper for cN, scNfM is a stopper for
 146         a feed in cN.  If no polling was started for that cN or cNfM,
 147         in its place is None.  """
 148         stoppers = []
 149         if pollcontext is None:
 150             pollcontext = {}
 151         for c in cats:
 152             category_stoppers = []
 153             sc = None
 154             if c.subscription is not None and not self._pollers.has_key(c) and not pollcontext.has_key(c):
 155                 self._pollers[c] = True
 156                 pollcontext[c] = True
 157                 sc = CategoryPoller(c, pollcontext).poll()
 158             category_stoppers.append(sc)
 159             if feeds:
 160                 for f in c.feeds:
 161                     sf = None
 162                     if not self._pollers.has_key(f) and not pollcontext.has_key(f):
 163                         self._pollers[f] = True
 164                         pollcontext[f] = True
 165                         sf = FeedPoller(f).poll(), f
 166                     category_stoppers.append(sf)
 167             stoppers.append(category_stoppers)
 168         return stoppers
 169
 170     def poll_done(self, obj):
 171         if self._pollers.has_key(obj):
 172             del self._pollers[obj]
 173
 174 class FeedPoller:
 175     def __init__(self, feed):
 176         self._feed = feed
 177
 178     def poll(self):
 179         MainloopManager.get_instance().call_pending()
 180
 181         url, user, pw = self._feed.access_info
 182         parsed = None
 183         config = Config.get_instance()
 184         self._feed.last_poll = int (time.time())
 185         ps = None
 186         try:
 187             try:
 188                 stopper = URLFetch.get_instance().request(
 189                     url, self, user, pw, priority=NetworkConstants.PRIORITY_RSS)
 190                 ps = PollStopper(stopper, self._feed)
 191             except Exception, e:
 192                 error.log_exc("Caught an exception while polling")
 193                 self.http_failed(e)
 194             else:
 195                 self._feed.router.start_polling()
 196                 self._feed.router.stopper = ps
 197         finally:
 198             return ps
 199
 200     def http_results(self, status, data):
 201         MainloopManager.get_instance().call_pending()
 202         try:
 203             try:
 204                 err = ""
 205                 read_data = True
 206                 #if status is None:
 207                 #    err = _("No data")
 208                 #elif status[1] == 304:
 209                 #    self._feed.router.route_no_data()
 210                 #    read_data = False
 211                 #elif status[1] == 410 or status[1] == 404:
 212                 #    err = _("Unable to find the feed (%s: %s)") % (
 213                 #        status[1], status[2].strip())
 214                 #elif status[1] == 401:
 215                 #    err = _("Invalid username and password.")
 216                 #elif status[1] > 299:
 217                 #    err = _("Updating feed resulted in abnormal status '%s' (code %d)") % (
 218                 #        status[2].strip(), status[1])
 219
 220                 (x, code, message) = status
 221                 if code == 304:
 222                     self._feed.router.route_no_data()
 223                     read_data = False
 224                 elif code > 299:
 225                     err = "%s" % message
 226
 227                 if err:
 228                     self._feed.router.set_error(err)
 229                 elif read_data:
 230                     try:
 231                         parsed = SummaryParser.parse(data, self._feed)
 232                     except Exception, e:
 233                         error.log_exc("exception in summaryparser")
 234                         self._feed.router.set_error(
 235                             _("An error occurred while processing feed: %s") %
 236                             str(e))
 237                     else:
 238                         self._feed.router.route_all(parsed)
 239                 post_status_message(_("Updating %s done.") % self._feed.title)
 240             except Exception, ex:
 241                 error.log_exc("error while parsing results")
 242                 self._feed.router.set_error(str(ex))
 243         finally:
 244             get_instance().poll_done(self._feed)
 245
 246     def http_failed(self, exception):
 247         try:
 248             if isinstance(exception, socket.error):
 249                 self._feed.router.route_no_data()
 250             else:
 251                 self._feed.router.set_error(str(exception))
 252             post_status_message(_("Updating %s failed") % self._feed.title)
 253             self._feed.poll_done()
 254         finally:
 255             get_instance().poll_done(self._feed)
 256
 257     def http_permanent_redirect(self, location):
 258         # XXX
 259         (oldloc, u, p) = self._feed.access_info
 260         self._feed.access_info = (location, u, p)
 261
 262     def operation_stopped(self):
 263         self._feed.router.route_no_data()
 264         self._feed.poll_done()
 265         get_instance().poll_done(self._feed)
 266
 267 class CategoryPoller:
 268     def __init__(self, category, pollcontext):
 269         self._category = category
 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 = helpers.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