Process NOTIFY events in reception order.

wulczer · wulczer · commit 66cd4670274f · 2014-06-23T14:23:53.000+02:00
Repeatedly popping them from the notify list caused observer functions to be called in reverse order. Instead, iterate over the list directly and clear it out at the and. Noticed by Max Walton, though I didn't use his patch. Fixes #31.
diff --git a/NEWS b/NEWS
@@ -5,6 +5,8 @@ What's new in txpostgres 1.3.0
     (while still prefering straingh psycopg2)
   - allow choosing the exact psycopg2 implementation using an
     environment variable
+  - NOTIFY events are now processed in reception order (thanks to
+    Max Walton)
 
 What's new in txpostgres 1.2.0
 ------------------------------
diff --git a/test/test_txpostgres.py b/test/test_txpostgres.py
@@ -1278,6 +1278,43 @@ def observer(notify):
         return d.addCallback(lambda _: self.assertEquals(
                 len(self.notifies), 3))
 
+    def test_notifyDeliveryOrder(self):
+        """
+        Notifications are delivered to observers in the same order that they
+        were sent.
+        """
+        # this tests accesses NOTIFY payloads, so it requires PostgreSQL 9.0+
+        # and Psycopg 2.3+
+        if self.conn.server_version < 90000:
+            raise unittest.SkipTest(
+                "PostgreSQL < 9.0.0 does not support NOTIFY payloads")
+
+        if getattr(psycopg2.extensions.Notify, 'payload', None) is None:
+            raise unittest.SkipTest(
+                "psycopg2 does not have NOTIFY payload support. You need at "
+                "least version 2.3.0 of psycopg2 to process NOTIFY payloads.")
+
+        dl = [defer.Deferred() for _ in range(10)]
+        payloads = map(str, range(10))
+        notifyD = defer.DeferredList(dl)
+
+        def observer(notify):
+            self.notifies.append(notify)
+            dl.pop().callback(None)
+
+        self.conn.addNotifyObserver(observer)
+
+        d = self.conn.runOperation("listen txpostgres_test")
+        # send all notification together to ensure that they are processed
+        # inside a single checkForNotifies call
+        d.addCallback(lambda _: self.notifyconn.runOperation(
+            'notify txpostgres_test, %s; ' * 10, payloads))
+        # wait for all notification to be processed
+        d.addCallback(lambda _: notifyD)
+        # check that they were processed in the right order
+        return d.addCallback(lambda _: self.assertEquals(
+                [n.payload for n in self.notifies], payloads))
+
     def test_multipleObservers(self):
         """
         Multiple registered notify observers each get notified.
diff --git a/txpostgres/txpostgres.py b/txpostgres/txpostgres.py
@@ -695,8 +695,7 @@ def checkForNotifies(self):
             self.cooperator.cooperate(self._checkForNotifies())
 
     def _checkForNotifies(self):
-        while self._connection.notifies:
-            notify = self._connection.notifies.pop()
+        for notify in self._connection.notifies:
             # don't iterate over self._notifyObservers directly because the
             # observer function might call removeNotifyObserver, thus modifying
             # the set while it's being iterated
@@ -708,6 +707,8 @@ def _checkForNotifies(self):
                 # observers
                 yield defer.maybeDeferred(observer, notify).addErrback(log.err)
 
+        del self._connection.notifies[:]
+
     def addNotifyObserver(self, observer):
         """
         Add an observer function that will get called whenever a :pg:`NOTIFY
