Some PyFlakes/PEP8 fixes (in the wrong place); threading in PyGTK

mitterlib/network/networkbase.py

@@ -24,7 +24,6 @@ The :mod:`networkbase` module defines the base classes for all networks.
 """
 
 import logging
-import datetime
 
 from mitterlib.constants import version
 

mitterlib/network/twitter.py

@@ -24,13 +24,15 @@ import datetime
 import base64
 import htmlentitydefs
 import re
+import warnings
 
 from httplib import BadStatusLine
 from socket import error as socketError
 
 from networkbase import NetworkBase, NetworkData, auth_options, \
         NetworkDNSError, NetworkBadStatusLineError, NetworkLowLevelError, \
-        NetworkInvalidResponseError, NetworkPermissionDeniedError
+        NetworkInvalidResponseError, NetworkPermissionDeniedError, \
+        MessageTooLongWarning
 
 try:
     # Python 2.6/3.0 JSON parser

mitterlib/ui/ui_cmd.py

@@ -27,7 +27,7 @@ import mitterlib.constants
 
 from mitterlib.network import NetworksNoNetworkSetupError, NetworksError
 from mitterlib.network.networkbase import NetworkError, \
-    NetworkPermissionDeniedError
+    NetworkPermissionDeniedError, MessageTooLongWarning
 
 _log = logging.getLogger('ui.cmd')
 

mitterlib/ui/ui_pygtk.py

@@ -25,18 +25,22 @@ gobject.threads_init()
 import logging
 import threading
 import Queue
+import re
 
 from mitterlib.ui.helpers.image_helpers import find_image
 
-#from mitterlib.constants import gpl_3, version
+from mitterlib.constants import gpl_3, version
 #from mitterlib.ui.helpers.utils import str_len
 #from mitterlib.ui.helpers.notify import Notify
-#from mitterlib.ui.helpers import timesince
+from mitterlib.ui.helpers import timesince
 
 # Constants
 
 _log = logging.getLogger('ui.pygtk')
 
+URL_RE = re.compile(
+        r'((?:(?:https?|ftp)://|www[-\w]*\.)[^\s\n\r]+[-\w+&@#%=~])', re.I)
+
 # ----------------------------------------------------------------------
 # Helper Functions (not related to objects or that don't need direct access to
 # the objects contents.)
@@ -75,65 +79,75 @@ class _WorkerThread(threading.Thread, _IdleObject):
     """
     __gsignals__ = {
             "completed": (
-                gobject.SIGNAL_RUN_LAST, gobject.TYPE_NONE, [
-                    gobject.TYPE_PYOBJECT]),    # The list/NetworkData object
+                gobject.SIGNAL_RUN_LAST,
+                gobject.TYPE_NONE,
+                (gobject.TYPE_PYOBJECT,)),  # list/networkdata
             "exception": (
-                gobject.SIGNAL_RUN_LAST, gobject.TYPE_NONE, [
-                    gobject.TYPE_PYOBJECT])     # The exception
+                gobject.SIGNAL_RUN_LAST,
+                gobject.TYPE_NONE, 
+                (gobject.TYPE_PYOBJECT,))   # The exception
             }
  
-    def __init__(self, queue):
+    def __init__(self, function, *args, **kwargs):
         threading.Thread.__init__(self)
         _IdleObject.__init__(self)
-        self._queue = queue
+        self._function = function
+        self._args = args
+        self._kwargs = kwargs
  
     def run(self):
-        while True:
-            _log.debug('Thread %d waiting for work', self.get_ident())
-            work = self._queue.get()
-            if work is None:
-                # End the thread
-                break
-
         # call the function
-            _log.debug('Thread %s got work', self.get_ident())
-            (complete, exception, function, args, kwargs) = work
+        _log.debug('Thread %s calling %s', self.ident, str(self._function))
 
-            _log.debug('Calling %s', str(function))
+        args = self._args
+        kwargs = self._kwargs
 
         try:
-                result = function(*args, **kwargs)
+            result = self._function(*args, **kwargs)
         except Exception, exc:  # Catch ALL exceptions
             # XXX: Check if this catch all warnins too!
             _log.debug('Exception %s', str(exc))
             self.emit("exception", exc)
 
-            self.emit("completed", result)
-
-            # work complete
-            self._queue.task_done()
+        _log.debug('Thread %s completed', self.name)
 
-        _log.debug('Thread %d ending', self.get_ident())
+        self.emit("completed", result)
         return
 
 
 class _ThreadManager(object):
     """Manages the threads."""
 
-    def __init__(self, pool_size=2):
+    def __init__(self, max_threads=2):
         """Start the thread pool. The number of threads in the pool is defined
         by `pool_size`, defaults to 2."""
-        self._queue = Queue.Queue()
+        self._max_threads = max_threads
         self._thread_pool = []
+        self._running = []
+        self._thread_id = 0
+
+        return
 
-        while pool_size > 0:
-            _log.debug('Starting thread %d', pool_size)
-            new_thread = _WorkerThread(self._queue)
-            new_thread.start()
-            self._thread_pool.append(new_thread)
-            pool_size -= 1
+    def _remove_thread(self, widget, arg=None):
+        """Called when the thread completes. We remove it from the thread list
+        (dictionary, actually) and start the next thread (if there is one)."""
+
+        # not actually a widget. It's the object that emitted the signal, in
+        # this case, the _WorkerThread object.
+        thread_id = widget.name 
+        
+        _log.debug('Thread %s completed, %d threads in the queue', thread_id,
+                len(self._thread_pool))
+
+        self._running.remove(thread_id)
+
+        if self._thread_pool:
+            if len(self._running) < self._max_threads:
+                next = self._thread_pool.pop()
+                _log.debug('Dequeuing thread %s', thread.name)
+                self._running.append(thread.name)
+                thread.start()
 
-        _log.debug('All threads started')
         return
 
     def add_work(self, complete_cb, exception_cb, func, *args, **kwargs):
@@ -144,9 +158,26 @@ class _ThreadManager(object):
         function to be called in the secondary threads. `args` and `kwargs`
         are parameters passed to the function."""
 
-        # wrap all the params in a nice tuple to be added to the queue.
-        work = (complete_cb, exception_cb, func, args, kwargs)
-        self._queue.put(work)
+        thread = _WorkerThread(func, *args, **kwargs)
+        thread_id = '%s' % (self._thread_id)
+
+        thread.connect('completed', complete_cb)
+        thread.connect('completed', self._remove_thread)
+        thread.connect('exception', exception_cb)
+        thread.setName(thread_id)
+
+        if len(self._running) < self._max_threads:
+            # immediatelly start the thread
+            self._running.append(thread_id)
+            thread.start()
+        else:
+            # add the thread to the queue
+            running_names = ', '.join(self._running)
+            _log.debug('Threads %s running, adding %s to the queue',
+                    running_names, thread_id)
+            self._thread_pool.append(thread)
+
+        self._thread_id += 1
         return
 
 # ----------------------------------------------------------------------
@@ -459,7 +490,7 @@ class Interface(object):
         message = re.sub(r'&(?!(amp;|gt;|lt;|quot;|apos;))', r'&amp;', message)
 
         # highlight URLs
-        message = url_re.sub(r'<span foreground="blue">\1</span>',
+        message = URL_RE.sub(r'<span foreground="blue">\1</span>',
                             message)
 
         # use a different highlight for the current user
@@ -521,6 +552,23 @@ class Interface(object):
         gtk.main_quit()
         return
 
+    # ------------------------------------------------------------
+    # Network related functions
+    # ------------------------------------------------------------
+
+    def _post_get_messages(self, widget, results):
+        """Function called after the data from the messages list is
+        retrieved."""
+        _log.debug('Got results')
+        #_log.debug('Results from the request: %s', str(results))
+        return
+
+    def _exception_get_messages(self, widget, exception):
+        """Function called if the retrival of current messages returns an
+        exception."""
+        _log.debug(str(exception))
+        return
+
     # ------------------------------------------------------------
     # Required functions for all interfaces
     # ------------------------------------------------------------
@@ -548,8 +596,6 @@ class Interface(object):
             self._default_pixmap = gtk.gdk.Pixbuf(gtk.gdk.COLORSPACE_RGB,
                     has_alpha=False, bits_per_sample=8, width=48, height=48)
 
-        self._main_window = self._create_main_window()
-
         #self._systray_setup()
 #        self.create_settings_dialog()
 #        self.username_field.set_text(default_username)
@@ -572,7 +618,14 @@ class Interface(object):
         """Call function; displays the interface. This method should appear on
         every interface."""
 
+        self._main_window = self._create_main_window()
         self._main_window.show_all()
+        self._threads = _ThreadManager()
+
+        # queue the first fetch
+        self._threads.add_work(self._post_get_messages, 
+                self._exception_get_messages,
+                self._connection.messages)
 
         gtk.main()
 

mitterlib/ui/ui_zork.py

@@ -19,16 +19,12 @@
 
 import logging
 import cmd
-import datetime
 import warnings
-import pickle
 
 import mitterlib.ui.helpers.console_utils as console_utils
-import mitterlib.constants
 
 from mitterlib.network import NetworksNoNetworkSetupError, NetworksError
-from mitterlib.network.networkbase import NetworkError, \
-    NetworkPermissionDeniedError
+from mitterlib.network.networkbase import NetworkError, MessageTooLongWarning
 
 _log = logging.getLogger('ui.cmd')
 

tests.py

@@ -48,7 +48,6 @@ class TwitterEncodingTests(unittest.TestCase):
     def test_twitter_unhtml(self):
         """Test the _unhtml() function inside the Twitter."""
 
-
         text = u'RT @fisl10: Abertas as inscrições para o ' \
                 'fisl10 http://tinyurl.com/cqrdsc'
         result = u'RT @fisl10: Abertas as inscrições para o fisl10 ' \