Added "ignore_queue" option to bypass message queue in emits

socketio/base_manager.py

@@ -113,7 +113,7 @@ class BaseManager(object):
         return r
 
     def emit(self, event, data, namespace, room=None, skip_sid=None,
-             callback=None):
+             callback=None, **kwargs):
         """Emit a message to a single client, a room, or all the clients
         connected to the namespace."""
         if namespace not in self.rooms or room not in self.rooms[namespace]:

socketio/pubsub_manager.py

@@ -37,7 +37,7 @@ class PubSubManager(BaseManager):
         self.server.logger.info(self.name + ' backend initialized.')
 
     def emit(self, event, data, namespace=None, room=None, skip_sid=None,
-             callback=None):
+             callback=None, **kwargs):
         """Emit a message to a single client, a room, or all the clients
         connected to the namespace.
 
@@ -46,6 +46,10 @@ class PubSubManager(BaseManager):
 
         The parameters are the same as in :meth:`.Server.emit`.
         """
+        if kwargs.get('ignore_queue'):
+            return super(PubSubManager, self).emit(
+                event, data, namespace=namespace, room=room, skip_sid=skip_sid,
+                callback=callback)
         namespace = namespace or '/'
         if callback is not None:
             if self.server is None:

socketio/server.py

@@ -177,7 +177,7 @@ class Server(object):
             namespace_handler
 
     def emit(self, event, data=None, room=None, skip_sid=None, namespace=None,
-             callback=None):
+             callback=None, **kwargs):
         """Emit a custom event to one or more connected clients.
 
         :param event: The event name. It can be any string. The event names
@@ -202,14 +202,22 @@ class Server(object):
                          that will be passed to the function are those provided
                          by the client. Callback functions can only be used
                          when addressing an individual client.
+        :param ignore_queue: Only used when a message queue is configured. If
+                             set to ``True``, the event is emitted to the
+                             clients directly, without going through the queue.
+                             This is more efficient, but only works when a
+                             single server process is used. It is recommended
+                             to always leave this parameter with its default
+                             value of ``False``.
         """
         namespace = namespace or '/'
         self.logger.info('emitting event "%s" to %s [%s]', event,
                          room or 'all', namespace)
-        self.manager.emit(event, data, namespace, room, skip_sid, callback)
+        self.manager.emit(event, data, namespace, room, skip_sid, callback,
+                          **kwargs)
 
     def send(self, data, room=None, skip_sid=None, namespace=None,
-             callback=None):
+             callback=None, **kwargs):
         """Send a message to one or more connected clients.
 
         This function emits an event with the name ``'message'``. Use
@@ -234,8 +242,16 @@ class Server(object):
                          that will be passed to the function are those provided
                          by the client. Callback functions can only be used
                          when addressing an individual client.
+        :param ignore_queue: Only used when a message queue is configured. If
+                             set to ``True``, the event is emitted to the
+                             clients directly, without going through the queue.
+                             This is more efficient, but only works when a
+                             single server process is used. It is recommended
+                             to always leave this parameter with its default
+                             value of ``False``.
         """
-        self.emit('message', data, room, skip_sid, namespace, callback)
+        self.emit('message', data, room, skip_sid, namespace, callback,
+                  **kwargs)
 
     def enter_room(self, sid, room, namespace=None):
         """Enter a room.

tests/test_pubsub_manager.py

@@ -85,6 +85,14 @@ class TestBaseManager(unittest.TestCase):
             self.assertRaises(ValueError, self.pm.emit, 'foo', 'bar',
                               callback='cb')
 
+    def test_emit_with_ignore_queue(self):
+        self.pm.connect('123', '/')
+        self.pm.emit('foo', 'bar', room='123', namespace='/',
+                     ignore_queue=True)
+        self.pm._publish.assert_not_called()
+        self.pm.server._emit_internal.assert_called_once_with('123', 'foo',
+                                                              'bar', '/', None)
+
     def test_close_room(self):
         self.pm.close_room('foo')
         self.pm._publish.assert_called_once_with(