wsmsg 0.15

A simple websocket/msgpack messaging library that provides pub/sub and rpc patterns.

wsmsg Package

Overview

Websockets are used, and the server will accept all authenticated connections to an endpoint root (default '/'). Authentication is via an HTTP bearer token on initial connection. Each node attaches to an endpoint specified in the connection URI. For example: wss://<host>:<port>/<endpoint>. Channels receive (and queue) messages. Subscriptions bind an endpoint to one or more channels. Subjects can be used to filter messages in a subscription. Message delivery is at most once, with flow control and no guaranteed persistence.

Flow Diagram

Publish Channel  Routing     Endpoint 
-------+--------+-----------+-------------
      / Queue -> Forward \ / Queue -> Node
Node <  Queue -> Forward  X         / Node
      \ Queue -> Forward / \ Queue <  Node
                                    \ Node

Message Format

Messages are sent/received as a msgpack array

• uid: bytes
  Binary-safe unique id of message - recommand ULID as bytes
• channel: str
  Channel this message will be sent to
• subject: str
  Subject of message
• reference: bytes
  Binary-safe id of message that this references
• data: any
  Subject-defined message data

Python

Message = NamedTuple('Message', uid=bytes, channel=str, subject=str, reference=bytes, data=any)

class Message(NamedTuple):
    uid: bytes
    channel: str
    subject: str
    reference: bytes
    data: any
    def __repr__(self):
        elements = [
            str(ulid.from_bytes(self.uid)),
            self.channel,
            self.subject,
            str(ulid.from_bytes(self.reference)) if self.reference else None,
            repr(self.data)
        ]
        return "Message(%s)" % ', '.join(elements)

System Messages

The 'system' channel has ultimate priority (100) over all other channels.
If there are any queued messages in the system queue, no other channels will forward.

Commands

• ack
  Acknowledge message receipt, not forwarded.
  Ack'd message uid should be in the refercence field.
  Next message for connection will not be dequeued or sent
  until previous message was ackowledged.
• subcribe
  • channel (endpoint: str, channel_pattern: str, subject_pattern: str)
    Subscribe endpoint to channel pattern with subject pattern filter.
  • message (endpoint: str, message_uid: bytes, expires: int)
    Subscribe endpoint to any messages that reference the message uid until the expiration (unix timestamp).
• unsubscribe (endpoint: str, channel_pattern: str)
  • channel (endpoint: str, channel_pattern: str)
    Unsubscribe endpoint from patterns matching pattern.
    Includes any subject filters.
  • message (endpoint: str, message_uid: bytes)
    Unsubscribe endpoint to any messages that reference the message uid.
• pause
  Pause all message transmissions.
  Messages can still be received and system commands processed.
• resume
  Resume message transmissions.
• stop
  Cleanly stop and shutdown server.
  Pending messages are persisted.
• channel
  • create (channel: str)
    Create channel queue.
  • size (channel: str, size: int)
    Set max size of channel queue. Old messages are dropped when full
  • priority (channel: str, priority: int)
    Set channel priority. Higher is better, range 0-99. Default of zero.
• endpoint
  • create (endpoint: str)
    Create endpoint queue
  • size (endpoint: str, size: int)
    Set max size of endpoint queue. Old messages are dropped when full

System Events

• stopping Server is shutting down (in one second), no more messages will be delivered
• node
  • joined (endpoint, uid)
  • left (endpoint, uid)
• endpoint
  • created (endpoint)
  • sized (channel: str)
• channel
  • created (channel: str)
  • sized (channel: str)
  • prioritized (channel: str)

Misc

Router

• Until stop
  • Wait for next message
  • For each subscriber
    • Push message to node Queue

Route Map?

class Route NamedTuple('Route', router=Router, dest=set[UUID])
RouteMap = dict[UUID, Route]

routes: RouteMap = 
{
    <channel>: <route>, 
    ...
}

while True:
    msg = channel.get()
    for node in routes[channel].dest:
      forward_message(msg, node)

Examples

Task Queue

By connecting all nodes to one endpoint, each node will get a fair shair of the messages in the 'order' channel with the subject 'submitted'.

All nodes connects to ws://<host>:<port>/group1
node1 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, ('group1', 'orders', 'submitted'))
node2 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, ('group1', 'orders', 'submitted'))
node3 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, ('group1', 'orders', 'submitted'))

Fanout

By using one endpoint per node, each node will get a copy of the messages in the 'order' channel with the subject 'submitted'.

Each node connects to ws://<host>:<port>/<node id>
node1 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<node id>, 'orders', 'submitted'))
node2 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<node id>, 'orders', 'submitted'))
node3 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<node id>, 'orders', 'submitted'))

Direct Delivery

If each node subscribes to a unique addressable channel, messages can be routed to each node by the sender.
To send a message to a single node, use channel 'node.' with any subject.
To send a message to all nodes, additionally subscribe all nodes to a 'nodes' channel and send to that.

Each node connects to ws://<host>:<port>/<node id>
node1 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<node id>, 'node.<node id>', '*'))
node2 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<node id>, 'node.<node id>', '*'))
node3 -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<node id>, 'node.<node id>', '*'))

RPC

Advantages: Less messages to send, supports heavy RPC traffic through server.
Disadvantages: Need to know what reply channel and subjects will be used, large number of channel subscriptions can slow down message forwarding.

Connect to ws://<host>:<port>/<id>
node -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<id>, 'order', 'created'))
node -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<id>, 'order', 'exception'))
...
# Generate uid
Register Queue with RX task/thread for reference=uid
node -> Message(uid, 'order', 'submitted', None, None, Check)
...
node <- Message(<message uid>, 'order', 'created', uid, None, <check uid>)
RX task/thread matches and puts message(s) in Queue
...
Get result(s) with timeout from Queue
Resend if needed or show error on exception
Unregister Queue from RX task/thread

RPC - Streamlined

Advantages: message subscription matching is much faster, RTT should be lower. No need to know what reply subject may be.
Disadvantages: more messages to send, total throughput may be lower if heavy use of RPC.
Notes: If you set the message subscription expiration, no need to unsubscribe.

server -> Message(ulid.new().bytes, 'system', 'subscribe.channel', None, (<endpoint>, 'rpc', 'order.*'))
server -> Message(ulid.new().bytes, 'system', 'channel.priority', None, ('rpc', 99))
server -> Message(ulid.new().bytes, 'system', 'endpoint.size', None, (<endpoint>, 10))
...
Connect to ws://<host>:<port>/<id>
...
Generate uid
Register Queue with RX task/thread for reference=uid
node -> Message(ulid.new().bytes, 'system', 'subscribe.message', None, (<id>, uid, <timeout>))
node -> Message(uid, 'rpc', 'order.get.request', None, Check)
...
node <- Message(<message uid>, 'rpc', 'order.get.reply', uid, None, <check uid>)
RX task/thread matches and puts message(s) in Queue
...
Get result(s) with timeout from Queue
Resend if needed or show error on exception
Unregister Queue from RX task/thread