[CalendarServer-changes] [8516] CalendarServer/trunk/calendarserver/push

[source_changes at macosforge.org]

Revision: 8516
          http://trac.macosforge.org/projects/calendarserver/changeset/8516
Author:   sagen at apple.com
Date:     2012-01-12 15:50:36 -0800 (Thu, 12 Jan 2012)
Log Message:
-----------
Adding better message handling to APNProvider.dataReceived( )

Modified Paths:
--------------
    CalendarServer/trunk/calendarserver/push/applepush.py
    CalendarServer/trunk/calendarserver/push/test/test_applepush.py

Modified: CalendarServer/trunk/calendarserver/push/applepush.py
===================================================================
--- CalendarServer/trunk/calendarserver/push/applepush.py	2012-01-12 18:25:41 UTC (rev 8515)
+++ CalendarServer/trunk/calendarserver/push/applepush.py	2012-01-12 23:50:36 UTC (rev 8516)
@@ -199,6 +199,8 @@
         255 : "None (unknown)",
     }
 
+    MESSAGE_LENGTH = 6
+
     def makeConnection(self, transport):
         self.identifier = 0
         # self.log_debug("ProviderProtocol makeConnection")
@@ -206,6 +208,7 @@
 
     def connectionMade(self):
         self.log_debug("ProviderProtocol connectionMade")
+        self.buffer = ""
         # Store a reference to ourself on the factory so the service can
         # later call us
         self.factory.connection = self
@@ -217,11 +220,30 @@
         self.factory.connection = None
 
     def dataReceived(self, data, fn=None):
+        """
+        Buffer and divide up received data into error messages which are
+        always 6 bytes long
+        """
+
+        if fn is None:
+            fn = self.processError
+
         self.log_debug("ProviderProtocol dataReceived %d bytes" % (len(data),))
-        command, status, identifier = struct.unpack("!BBI", data)
-        if command == self.COMMAND_ERROR:
-            self.processError(status, identifier)
+        self.buffer += data
 
+        while len(self.buffer) >= self.MESSAGE_LENGTH:
+            message = self.buffer[:self.MESSAGE_LENGTH]
+            self.buffer = self.buffer[self.MESSAGE_LENGTH:]
+
+            try:
+                command, status, identifier = struct.unpack("!BBI", message)
+                if command == self.COMMAND_ERROR:
+                    fn(status, identifier)
+            except Exception, e:
+                self.log_warn("ProviderProtocol could not process error: %s (%s)" %
+                    (message.encode("hex"), e))
+
+
     def processError(self, status, identifier):
         """
         Handles an error message we've received from on feedback channel.

Modified: CalendarServer/trunk/calendarserver/push/test/test_applepush.py
===================================================================
--- CalendarServer/trunk/calendarserver/push/test/test_applepush.py	2012-01-12 18:25:41 UTC (rev 8515)
+++ CalendarServer/trunk/calendarserver/push/test/test_applepush.py	2012-01-12 23:50:36 UTC (rev 8516)
@@ -108,11 +108,37 @@
             rawData[45:])
         self.assertEquals(payload[0], '{"key" : "%s"}' % (key1,))
 
+        def errorTestFunction(status, identifier):
+            history.append((status, identifier))
+            return succeed(None)
+
         # Simulate an error
-        errorData = struct.pack("!BBI", APNProviderProtocol.COMMAND_ERROR, 1, 1)
-        yield connector.receiveData(errorData)
+        history = []
+        errorData = struct.pack("!BBI", APNProviderProtocol.COMMAND_ERROR, 1, 2)
+        yield connector.receiveData(errorData, fn=errorTestFunction)
         clock.advance(301)
 
+        # Simulate multiple errors and dataReceived called
+        # with amounts of data not fitting message boundaries
+        # Send 1st 4 bytes
+        history = []
+        errorData = struct.pack("!BBIBBI",
+            APNProviderProtocol.COMMAND_ERROR, 3, 4,
+            APNProviderProtocol.COMMAND_ERROR, 5, 6,
+        )
+        yield connector.receiveData(errorData[:4], fn=errorTestFunction)
+        # Send remaining bytes
+        yield connector.receiveData(errorData[4:], fn=errorTestFunction)
+        self.assertEquals(history, [(3, 4), (5, 6)])
+        # Buffer is empty
+        self.assertEquals(len(connector.service.protocol.buffer), 0)
+
+        # Sending 7 bytes
+        yield connector.receiveData("!" * 7, fn=errorTestFunction)
+        # Buffer has 1 byte remaining
+        self.assertEquals(len(connector.service.protocol.buffer), 1)
+
+
         # Prior to feedback, there are 2 subscriptions
         txn = self.store.newTransaction()
         subscriptions = (yield txn.apnSubscriptionsByToken(token))
@@ -131,7 +157,7 @@
         # Simulate feedback with multiple tokens, and dataReceived called
         # with amounts of data not fitting message boundaries
         history = []
-        def testFunction(timestamp, token):
+        def feedbackTestFunction(timestamp, token):
             history.append((timestamp, token))
             return succeed(None)
         timestamp = 2000
@@ -141,15 +167,15 @@
             timestamp, len(binaryToken), binaryToken,
             )
         # Send 1st 10 bytes
-        yield connector.receiveData(feedbackData[:10], fn=testFunction)
+        yield connector.receiveData(feedbackData[:10], fn=feedbackTestFunction)
         # Send remaining bytes
-        yield connector.receiveData(feedbackData[10:], fn=testFunction)
+        yield connector.receiveData(feedbackData[10:], fn=feedbackTestFunction)
         self.assertEquals(history, [(timestamp, token), (timestamp, token)])
         # Buffer is empty
         self.assertEquals(len(connector.service.protocol.buffer), 0)
 
         # Sending 39 bytes
-        yield connector.receiveData("!" * 39, fn=testFunction)
+        yield connector.receiveData("!" * 39, fn=feedbackTestFunction)
         # Buffer has 1 byte remaining
         self.assertEquals(len(connector.service.protocol.buffer), 1)
 
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20120112/40601e42/attachment-0001.html>