"""Manage a client list attached to a pub/sub backend.

name = 'pubsub'

def __init__(self, channel='socketio', write_only=False, logger=None):

super(PubSubManager, self).__init__()

self.channel = channel

self.write_only = write_only

self.host_id = uuid.uuid4().hex

self.logger = logger

def initialize(self):

super(PubSubManager, self).initialize()

if not self.write_only:

self.thread = self.server.start_background_task(self._thread)

self._get_logger().info(self.name + ' backend initialized.')

def emit(self, event, data, namespace=None, room=None, skip_sid=None,

callback=None, **kwargs):

"""Emit a message to a single client, a room, or all the clients

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:

raise RuntimeError('Callbacks can only be issued from the '

'context of a server.')

if room is None:

raise ValueError('Cannot use callback without a room set.')

id = self._generate_ack_id(room, namespace, callback)

callback = (room, namespace, id)

else:

callback = None

self._publish({'method': 'emit', 'event': event, 'data': data,

'namespace': namespace, 'room': room,

'skip_sid': skip_sid, 'callback': callback,

'host_id': self.host_id})

def can_disconnect(self, sid, namespace):

if self.is_connected(sid, namespace):

# client is in this server, so we can disconnect directly

return super().can_disconnect(sid, namespace)

else:

# client is in another server, so we post request to the queue

self._publish({'method': 'disconnect', 'sid': sid,

'namespace': namespace or '/'})

def close_room(self, room, namespace=None):

self._publish({'method': 'close_room', 'room': room,

'namespace': namespace or '/'})

def _publish(self, data):

"""Publish a message on the Socket.IO channel.

raise NotImplementedError('This method must be implemented in a '

'subclass.') # pragma: no cover

def _listen(self):

"""Return the next message published on the Socket.IO channel,

raise NotImplementedError('This method must be implemented in a '

'subclass.') # pragma: no cover

def _handle_emit(self, message):

# Events with callbacks are very tricky to handle across hosts

# Here in the receiving end we set up a local callback that preserves

# the callback host and id from the sender

remote_callback = message.get('callback')

remote_host_id = message.get('host_id')

if remote_callback is not None and len(remote_callback) == 3:

callback = partial(self._return_callback, remote_host_id,

*remote_callback)

else:

callback = None

super(PubSubManager, self).emit(message['event'], message['data'],

namespace=message.get('namespace'),

room=message.get('room'),

skip_sid=message.get('skip_sid'),

callback=callback)

def _handle_callback(self, message):

if self.host_id == message.get('host_id'):

try:

sid = message['sid']

namespace = message['namespace']

id = message['id']

args = message['args']

except KeyError:

return

self.trigger_callback(sid, namespace, id, args)

def _return_callback(self, host_id, sid, namespace, callback_id, *args):

# When an event callback is received, the callback is returned back

# the sender, which is identified by the host_id

self._publish({'method': 'callback', 'host_id': host_id,

'sid': sid, 'namespace': namespace, 'id': callback_id,

'args': args})

def _handle_disconnect(self, message):

self.server.disconnect(sid=message.get('sid'),

namespace=message.get('namespace'),

ignore_queue=True)

def _handle_close_room(self, message):

super(PubSubManager, self).close_room(

room=message.get('room'), namespace=message.get('namespace'))

def _thread(self):

for message in self._listen():

data = None

if isinstance(message, dict):

data = message

else:

if isinstance(message, six.binary_type): # pragma: no cover

try:

data = pickle.loads(message)

except:

pass

if data is None:

try:

data = json.loads(message)

except:

pass

if data and 'method' in data:

self._get_logger().info('pubsub message: {}'.format(

data['method']))

if data['method'] == 'emit':

self._handle_emit(data)

elif data['method'] == 'callback':

self._handle_callback(data)

elif data['method'] == 'disconnect':

self._handle_disconnect(data)

elif data['method'] == 'close_room':