src/lib/URLFetch.py

   1 """ URLFetch_threaded.py
   2
   3 Module for retrieving data from a URL (a threaded version using urllib2).
   4
   5 """
   6 __copyright__ = "Copyright (c) 2007 Straw developers"
   7 __license__ = """
   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 from MainloopManager import schedule
  22 import NetworkConstants
  23 import constants
  24 import Config
  25
  26 import threading
  27 import urllib2
  28
  29 import httplib2
  30
  31 from sys import exc_info
  32 import os, os.path
  33
  34 class CancelledException(Exception):
  35     """Operation is cancelled."""
  36     pass
  37
  38 class ConnectionThread(threading.Thread):
  39     """A thread that fetches a URL XXX fetch several URLs"""
  40     def __init__(self, uri, consumer, headers, client_object, group=None, target=None, name=None, *args, **kwargs):
  41         threading.Thread.__init__(self, group, target, name, args, kwargs)
  42         self.setDaemon(True)
  43
  44         self._uri = uri
  45         self._consumer = consumer
  46         self._headers = headers
  47         self._httpclient = client_object
  48
  49         self._cancelled = threading.Event()
  50
  51     def run(self):
  52         """The main loop of the thread"""
  53         print "Fetching %s..." % self._uri
  54         try:
  55             self._handle_request()
  56         finally:
  57             schedule(get_instance().request_done, self)
  58             print "Finished with %s" % self._uri
  59
  60     def cooperate(self):
  61         """
  62         This should be called periodically in the thread execution.
  63         The method checks whether cancellation has been requested
  64         and if so, raises CancelledException.
  65         """
  66         if self._cancelled.isSet():
  67             raise CancelledException
  68
  69     def _handle_request(self):
  70         data = None
  71         try:
  72             self.cooperate()
  73             try:
  74                 (response, data) = self._httpclient.request(self._uri, headers=self._headers)
  75             except httplib2.HttpLib2Error, exception:
  76                 if hasattr(exception, 'response'):
  77                     response = exception.response
  78                 elif hasattr(exception, 'content'):
  79                     data = exception.content
  80                 else:
  81                     raise
  82         except CancelledException:
  83             schedule(self._consumer.operation_stopped)
  84         except:
  85             try:
  86                 self.cooperate()  # last chance to notice cancellation
  87             except CancelledException:
  88                 schedule(self._consumer.operation_stopped)
  89             else:
  90                 schedule(self._consumer.http_failed, exc_info()[1])
  91         else:
  92             if not hasattr(response, 'status'):  # fake for non-http
  93                 f.status = 200
  94                 f.reason = "OK"
  95             schedule(self._consumer.http_results, (None, response.status, response.reason), data)
  96
  97     def cancel(self):
  98         """
  99         This can be called to cancel the request.
 100         inter-thread safe but not instant
 101         XXX network operations can take a long time to timeout
 102         XXX call operation_stopped instantly?
 103         """
 104         self._cancelled.set()
 105
 106 class ConnectionManager:
 107     """A manager for threads that fetch URLs"""
 108
 109     CACHE_DIR = os.path.join(Config.straw_home(), 'cache')
 110
 111     def __init__(self):
 112         self._starting = []  # requests waiting to be started
 113         self._active = []    # requests running right now
 114
 115     def request(self, uri, consumer, headers={}, user=None, password=None, priority=NetworkConstants.PRIORITY_DEFAULT):
 116
 117         httpclient = httplib2.Http(ConnectionManager.CACHE_DIR)
 118
 119         if user and password:
 120             httpclient.add_credentials(user,password)
 121
 122         config = Config.get_instance()
 123         proxy = config.proxy
 124         if proxy.active:
 125             httpclient.set_proxy(proxy.host, proxy.port)
 126
 127         if not headers:
 128             headers = {}
 129         headers['user-agent'] = 'Straw/%s' % constants.VERSION
 130
 131         thread = ConnectionThread(uri, consumer, headers, httpclient)
 132
 133         if len(self._active) < NetworkConstants.MAX_CONNECTIONS:
 134             self._active.append(thread)
 135             thread.start()
 136         else:
 137             self._starting.append(thread)
 138
 139         return thread.cancel # inter-thread safe
 140
 141     def request_done(self, request):
 142         """Called by the request when it is finished."""
 143         self._active.remove(request)
 144         if self._starting and len(self._active) < NetworkConstants.MAX_CONNECTIONS:
 145             thread = self._starting.pop(0)
 146             self._active.append(thread)
 147             thread.start()
 148
 149 def create_instance():
 150     return ConnectionManager()
 151
 152 connection_manager_instance = None
 153 def get_instance():
 154     global connection_manager_instance
 155     if connection_manager_instance is None:
 156         connection_manager_instance = create_instance()
 157     return connection_manager_instance
 158
 159 if __name__ == '__main__':
 160     import sys, gobject
 161     gobject.threads_init()
 162
 163     class Stub:
 164         def http_results(self, status, data):
 165             print status
 166             print info
 167             print "%s bytes of content" % len(data)
 168         def http_permanent_redirect(self, location):
 169             print "Redirected to %s" % location
 170         def http_failed(self, e):
 171             print str(e)
 172         def operation_stopped(self):
 173             print "Operation stopped"
 174
 175     for uri in sys.argv[1:]:
 176         get_instance().request(uri, Stub())
 177
 178     try:
 179         gobject.MainLoop().run()
 180     finally:
 181         print get_instance()._queue