server.py

## Central Server

import asyncio
import websockets
from re import sub
from signal import signal, SIGINT
import json

##Subscribers for each publisher
subscribers = dict()
subscribers['this'] = ['test1', 'test2']
subscribers['that'] = ['test1', 'test2']
subscribers['candle'] = ['plotter1', 'plotter', 'strategy', 'logger']
subscribers['ticker'] = ['strategy', 'logger']
subscribers['strategy'] = ['trader', 'papertrader']
subscribers['trader'] = ['strategy', 'logger']
subscribers['papertrader'] = ['logger', 'portfolio']
subscribers['portfolio'] = ['logger']

clients = dict()

async def register(websocket, path):
    global clients
    # if path in clients.keys():
    #     print(path+' already registered')
    #     return
    
    clients[path] = websocket
    print('successfully registered '+path)

async def unregister(path):
    global clients
    try:
        clients.pop(path)
        print('successfully unregistered '+path)
    except:
        print(path+' is not registered')
    return

async def send_msg(path, msg):
    try:
        await clients[path].send(msg)
    except websockets.exceptions.ConnectionClosedError:
        await unregister(path)

async def hello(websocket, path):
    path = path[1:]
   
    await register(websocket, path)
    try:
        while True:
            if path in subscribers.keys():
                msg = await websocket.recv()
                a = [type(clients[i]) for i in subscribers[path] if i in clients.keys()]
                if len(a)>0:
                    await asyncio.wait([send_msg(i, path+'?'+msg) for i in subscribers[path] if i in clients.keys()])
            else:
                msg = await websocket.recv()
                print(msg)
                await asyncio.sleep(0.1)

    except websockets.exceptions.ConnectionClosedError:
        await unregister(path)
    except asyncio.exceptions.IncompleteReadError:
        print('incomplete error')


def handler(signal_received, frame):
    exit(0)

if __name__ == '__main__':
    start_server = websockets.serve(hello, "localhost", 8765)
    signal(SIGINT, handler)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(start_server)
    loop.run_forever()