Protocol¶

Basic messages protocol and design guide.

Messages are presented via JSON strings for sending through network and via standard dict objects for local dispatching.

Messages in Bachata can be of two types:

1. Data messages
2. Transport messages

{
    "id": (str) Message ID,
    "type": (str) Message type,
    "time": (int) Unix-style timestamp in milliseconds,
    "data": (str) / (obj) String or object,
    "from": (str, optional) Sender, may be empty for messages "by system"
    "dest": (str, optional) Destination, may be empty if it's non-adressed
    "sign": (str, optional) Signature
}

{
    "type": (int) Transport type,
    "data": (str) Message ID,
    "sign": (str, optional) Signature
}

Messages center¶

class bachata.BaseMessagesCenter(loop=None, proto=None, queue=None)[source]¶

Messages center provides top-level messages routing.

Parameters:	loop – asyncio event loop queue – Messages queue instance proto – Messages protocol instance or BaseProtocol instance will be used by default

Attributes:

• .proto: BaseProtocol or subclass instance
• .loop: asyncio event loop
• .queue: BaseQueue subclass instance
• .routes: routes objects list

add_route(route)[source]¶

Add messages route to routing chain, see BaseRoute

Message routes are processed in the same order as they added.

Parameters:	route – Route instance

add_socket(channel, websocket)[source]¶

Register WebSocket for receiving messages from channel.

Parameters:	channel – String channel identifier, based on user id or some hash string websocket – Tornado WebSocket handler instance

del_route(route)[source]¶

Remove message route, see BaseRoute

Parameters:	route – Route instance

del_socket(channel, websocket)[source]¶

Unregister WebSocket from receiving messages from channel.

Parameters:	channel – String channel identifier, based on user id or some hash string websocket – Tornado WebSocket handler instance

process(raw_or_message, websocket=None)[source]¶

Process message to routing chain and send to WebSockets.

Message is passed to registered routes, they return receivers channels and then message is sent to that channels.

Parameters:	raw_or_message – Raw message string or message dict websocket – WebSocket connection handler received new message, this is optional parameter, because message could also be created at server internally

transport(message=None, websocket=None)[source]¶

Process transport layer for messages.

Performs reliable delivery process, notifies sender on successful message delivery.

Parameters:	message – Data or transport message to process websocket – WebSocket which has received message to process

WebSocket handler¶

class bachata.tornado.MessagesHandler(application, request, **kwargs)[source]¶

Base WebSocket handler for Tornado.

Default implementation works with authentication free channels, it just generates random channel identifier.

Redefine get_channel() in subclass if you want to have channels identifiers based on authenticated user.

authenticate()[source]¶

Authenticate and load current user, asyncio coroutine.

Default implementation is empty, so real authentication logic should be implemented in subclass.

get_channel()[source]¶

Get channel string identifier for connection. Method must be defined in subclass.

Implementation example:

class MyMessagesHandler(talkie.tornado.MessagesHandler):
    def get_channel(self):
        if self.current_user:
            return 'channel:%s' % self.current_user.id

get_messages_center()[source]¶

Get messages center for WebSocket. Method must be defined in subclass.

Implementation example:

def get_messages_center(self):
    # bachata.RedisMessagesCenter instance
    return self.application.messages

on_auth_error()[source]¶

Close connection on authorization error.

on_close()[source]¶

Remove handler from messages center.

on_message(raw_message)[source]¶

Process message to messages center.

open()[source]¶

Open WebSocket connection and add socket channel to messages center.

It also performs authenticate() coroutine call, so get_channel() can rely on current authenticated user.

Redis stack¶

class bachata.redis.RedisMessagesCenter(loop=None, conn_params=None, reliable=False)[source]¶

Messages center on top of Redis LPUSH / BRPOP pattern.

After creating messages center instance init() coroutine also must be called.

Parameters:	loop – asyncio event loop conn_params – Redis connection params as dict reliable – Use reliable queue or simple queue, default is False

init()[source]¶

Setup main Redis connection for queue.

class bachata.redis.RedisQueue(loop=None, conn_params=None)[source]¶

Messages queue on top of Redis LPUSH / BRPOP pattern.

Schema description:

1. Messages are LPUSH’ed to list with “{channel}” key.
2. Receiver just listens for “{channel}” list updates with BRPOP.

Parameters:	loop – asyncio event loop websocket – WebSocket handler instance conn_params – Redis connection params

add_socket(channel, websocket, proto=None)[source]¶

Register WebSocket for receiving messages from channel.

Parameters:	channel – String channel identifier, based on user id or some hash string websocket – Tornado WebSocket handler instance proto – Messages protocol instance

connect()[source]¶

Setup main Redis connection.

del_socket(channel, websocket, proto=None)[source]¶

Unregister WebSocket from receiving messages from channel.

Parameters:	channel – String channel identifier, based on user id or some hash string websocket – Tornado WebSocket handler instance proto – Messages protocol instance

listen_queue(channel, websocket)[source]¶

Start queue listener for channel and WebSocket connection.

put_message(channels, message, proto=None, from_channel=None)[source]¶

Put messages on queue.

Parameters:	channels – List of destination channels message – Message dict object proto – Messages protocol instance from_channel – Message from channel