Handle exceptions broadcasting WS messages gracefully

If an exception occurs whilst processing an NSQ message
in the WebSocket server, log the exception and continue
on to the next message.

Since there is a single greenlet iterating over all
NSQ messages that might result in a WebSocket broadcast,
an error processing one message would previously cause all
subsequent WS broadcasts to cease.

h/streamer.py

@@ -563,10 +563,14 @@ def broadcast_from_queue(queue, sockets):
     appropriate active sessions.
     """
     for (topic, message) in queue:
-        if topic == ANNOTATIONS_TOPIC:
-            broadcast_annotation_message(message, sockets)
-        elif topic == USER_TOPIC:
-            broadcast_session_change_message(message, sockets)
+        try:
+            if topic == ANNOTATIONS_TOPIC:
+                broadcast_annotation_message(message, sockets)
+            elif topic == USER_TOPIC:
+                broadcast_session_change_message(message, sockets)
+        except Exception as e:
+            log.exception('Failed to process message "%s" from topic "%s"',
+                message, topic)
 
 
 def broadcast_annotation_message(message, sockets):

h/test/streamer_test.py

@@ -302,6 +302,15 @@ def test_terminated_socket_does_not_recieve_event(self):
         broadcast_from_queue(self.queue, [sock])
         assert sock.send.called is False
 
+    def test_send_handles_socket_error(self):
+        # if an error occurs whilst processing one queue message,
+        # processing of other queue messages should not be affected
+        self.should.return_value = True
+        sock = FakeSocket('aclient')
+        sock.send.side_effect = [IOError('socket error'), None, None]
+        broadcast_from_queue(self.queue, [sock])
+        assert sock.send.call_count == 3
+
 
 class TestBroadcastSessionChangeEvent(unittest.TestCase):
     def test_should_send_session_change_when_joining_or_leaving_group(self):