version 0.4.3

.travis.yml

@@ -9,7 +9,7 @@ python:
 install:
   - if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install --use-mirrors argparse unittest2; fi 
   - python setup.py install
-script: python -m runtests --profile --benchmark
+script: python -m runtests --profile --benchmark --verbosity 2
 
 branches:
   only:

CHANGELOG.rst

@@ -1,6 +1,7 @@
-Ver. 0.4.3
+Ver. 0.4.3 - 2012-Dec-28
 ==============================
-* Removed the loop tasks.
+* Removed the tasks in event loop. Task can only be added by appending callbacks or timeouts.
+* Fixed critical bug in MultiDeferred.
 * Test suite works on with multiple test workers.
 * Fixed issue #17 on asynchronous shell application.
 * Dining philosophers example works on events only.

pulsar/__init__.py

@@ -1,5 +1,5 @@
 '''Event driven concurrent framework for Python'''
-VERSION = (0, 4, 3, 'beta', 1)
+VERSION = (0, 4, 3, 'final', 1)
 
 from .utils.version import get_version
 

pulsar/async/defer.py

@@ -407,7 +407,6 @@ def callback(self, result=None):
     def result_or_self(self):
         '''Obtain the result if available, otherwise it returns self.'''
         return self.result if self._called and not self.paused else self
-        #return self.result if self._called else self
 
     ##################################################    INTERNAL METHODS
     def _run_callbacks(self):
@@ -619,9 +618,7 @@ def _add(self, key, value):
         value = maybe_async(value)
         if isinstance(value, (dict, list, tuple, set, frozenset)):
             value = self._make(value)    
-        if is_async(value):
-            self._add_deferred(key, value)
-        elif self.handle_value:
+        if not is_async(value) and self.handle_value:
             try:
                 val = self.handle_value(value)
             except Exception as e:
@@ -630,6 +627,9 @@ def _add(self, key, value):
                 if val is not value:
                     return self._add(key, val)
         self._setitem(key, value)
+        # add callback if an asynchronous value
+        if is_async(value):
+            self._add_deferred(key, value)
 
     def _make(self, value):
         md = self.__class__(value, fireOnOneErrback=self.fireOnOneErrback,

pulsar/async/iostream.py

@@ -501,18 +501,23 @@ class ClientSocketHandler(BaseSocketHandler):
     :class:`ClientSocketHandler`.
 '''
     parser_class = EchoParser
-    def __init__(self, socket, address, parser_class=None, timeout=None):
+    def __init__(self, socket, address, parser_class=None, timeout=None,
+                  read_timeout=None):
         '''Create a client or client-connection socket. A parser class
 is required in order to use :class:`SocketClient`.
 
 :parameter socket: a client or client-connection socket
 :parameter address: The address of the remote client/server
 :parameter parser_class: A class used for parsing messages.
 :parameter timeout: A timeout in seconds for the socket. Same rules as
-    the ``socket.settimeout`` method in the standard library.
+    the ``socket.settimeout`` method in the standard library. A value of 0
+    indicates an asynchronous socket.
+:parameter read_timeout: A timeout in seconds for asynchronous operations. This
+    value is only used when *timeout* is 0 in the constructor
+    of a :class:`AsyncIOStream`.
 '''
         self._socket_timeout = get_socket_timeout(timeout)
-        self._set_socket(socket)
+        self._set_socket(socket, read_timeout)
         self.remote_address = address
         parser_class = parser_class or self.parser_class
         self.parser = parser_class()
@@ -522,7 +527,7 @@ def __repr__(self):
         return str(self.remote_address)
     __str__ = __repr__
 
-    def _set_socket(self, sock):
+    def _set_socket(self, sock, read_timeout=None):
         if not isinstance(sock, AsyncIOStream):
             if self._socket_timeout == 0:
                 sock = AsyncIOStream(sock)
@@ -533,6 +538,7 @@ def _set_socket(self, sock):
         if self.async:
             close_callback = Deferred().add_callback(self.close)
             self.sock.set_close_callback(close_callback)
+            self.read_timeout = read_timeout
 
     def _get_read_timeout(self):
         if self.async:
@@ -544,7 +550,7 @@ def _set_read_timeout(self, value):
             self.sock._read_callback_timeout = value
         else:
             self._socket_timeout = value
-            sock.settimeout(self._socket_timeout)
+            self.sock.settimeout(self._socket_timeout)
     read_timeout = property(_get_read_timeout, _set_read_timeout) 
 
 
@@ -558,9 +564,10 @@ def __init__(self, *args, **kwargs):
         self.processing = False
 
     @classmethod
-    def connect(cls, address, parser_class=None, timeout=None):
+    def connect(cls, address, **kwargs):
+        '''Create a new :class:`ClientSocket` connected at *address*.'''
         sock = create_connection(address)
-        return cls(sock, address, parser_class=parser_class, timeout=timeout)
+        return cls(sock, address, **kwargs)
 
     def send(self, data):
         '''Send data to remote server'''

tests/iostream.py

@@ -71,6 +71,10 @@ def testSyncClient(self):
         client = self.client(timeout=3)
         self.assertFalse(client.async)
         self.assertEqual(client.gettimeout(), 3)
+        self.assertEqual(client.read_timeout, 3)
+        client.read_timeout = 4
+        self.assertEqual(client.gettimeout(), 4)
+        self.assertEqual(client.read_timeout, 4)
         self.assertEqual(client.execute(b'ciao'), b'ciao')
         self.assertEqual(client.received, 1)
         self.assertEqual(client.execute(b'bla'), b'bla')
@@ -162,7 +166,9 @@ def testMaxBufferSize(self):
         self.assertTrue(client.closed)
 
     def testConnectionClose(self):
-        client = self.client(timeout=0)
+        client = self.client(timeout=0, read_timeout=3)
+        self.assertTrue(client.async)
+        self.assertEqual(client.read_timeout, 3)
         future = client.execute(b'ciao')
         yield future
         self.assertEqual(future.result, b'ciao')