Replace dipatcher.pop by dispatcher.dispatch .

It is important that unregistering a response from dispatcher and putting
it in its queue are atomic at dispatcher level to support askStoreObject
pipelining.

git-svn-id: https://svn.erp5.org/repos/neo/trunk@1772 71dcc9de-d417-0410-9af5-da40c76e7ee4

neo/client/handlers/__init__.py

@@ -38,10 +38,8 @@ class BaseHandler(EventHandler):
     def packetReceived(self, conn, packet):
         """Redirect all received packet to dispatcher thread."""
         if packet.isResponse():
-            queue = self.dispatcher.pop(conn, packet.getId(), None)
-            if queue is None:
+            if not self.dispatcher.dispatch(conn, packet.getId(), (conn, packet)):
                 raise ProtocolError('Unexpected response packet')
-            queue.put((conn, packet))
         else:
             self.dispatch(conn, packet)
 

neo/dispatcher.py

@@ -16,7 +16,6 @@
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 
 from neo.locking import Lock
-MARKER = []
 EMPTY = {}
 
 def giant_lock(func):
@@ -38,12 +37,13 @@ class Dispatcher:
         self.lock_release = lock.release
 
     @giant_lock
-    def pop(self, conn, msg_id, default=MARKER):
-        """Retrieve register-time provided payload."""
-        result = self.message_table.get(id(conn), EMPTY).pop(msg_id, default)
-        if result is MARKER:
-            raise KeyError, (id(conn), msg_id)
-        return result
+    def dispatch(self, conn, msg_id, data):
+        """Retrieve register-time provided queue, and put data in it."""
+        queue = self.message_table.get(id(conn), EMPTY).pop(msg_id, None)
+        if queue is None:
+            return False
+        queue.put(data)
+        return True
 
     @giant_lock
     def register(self, conn, msg_id, queue):

neo/tests/testDispatcher.py

@@ -19,6 +19,7 @@ import unittest
 
 from mock import Mock
 from neo.dispatcher import Dispatcher
+from Queue import Queue
 
 class DispatcherTests(unittest.TestCase):
 
@@ -27,9 +28,15 @@ class DispatcherTests(unittest.TestCase):
 
     def testRegister(self):
         conn = object()
-        self.dispatcher.register(conn, 1, 0)
-        self.assertEqual(self.dispatcher.pop(conn, 1, None), 0)
-        self.assertEqual(self.dispatcher.pop(conn, 2, 3), 3)
+        queue = Queue()
+        MARKER = object()
+        self.dispatcher.register(conn, 1, queue)
+        self.assertTrue(queue.empty())
+        self.assertTrue(self.dispatcher.dispatch(conn, 1, MARKER))
+        self.assertFalse(queue.empty())
+        self.assertTrue(queue.get(block=False) is MARKER)
+        self.assertTrue(queue.empty())
+        self.assertFalse(self.dispatcher.dispatch(conn, 2, None))
 
     def testUnregister(self):
         conn = object()
@@ -37,7 +44,7 @@ class DispatcherTests(unittest.TestCase):
         self.dispatcher.register(conn, 2, queue)
         self.dispatcher.unregister(conn)
         self.assertEqual(len(queue.mockGetNamedCalls('put')), 1)
-        self.assertEqual(self.dispatcher.pop(conn, 2, 3), 3)
+        self.assertFalse(self.dispatcher.dispatch(conn, 2, None))
 
     def testRegistered(self):
         conn1 = object()