timesock.py

import asyncio
import websockets
import ssl
import json


async def producer():
    version = {'operation': 'version'}
    return json.dumps(version)


async def consumer(msg):
    print(msg)


async def handler(websocket):
    while True:
        listener_task = asyncio.ensure_future(websocket.recv())
        producer_task = asyncio.ensure_future(producer())
        done, pending = await asyncio.wait(
            [listener_task, producer_task],
            return_when=asyncio.FIRST_COMPLETED)

        if listener_task in done:
            message = listener_task.result()
            await consumer(message)
        else:
            listener_task.cancel()

        if producer_task in done:
            message = producer_task.result()
            await websocket.send(message)
        else:
            producer_task.cancel()


async def connect_timely():
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    async with websockets.connect("wss://localhost:54323/websocket", ssl=ctx) as websocket:
        version_op = json.dumps({'operation': 'version'})

        query = {
            "operation": "query",
            "sessionId": "13454051-ec86-4707-9510-023975cc4c05",
            "start": 0,
            "end":   1472737982000,
            "queries": [
                {
                    "aggregator": "max",
                    "metric": "sys.cpu.user",
                    "rate": "true",
                    "rateOptions": {
                        "counter": False,
                        "counterMax": 100,
                        "resetValue": 0
                    },
                    "downsample": "1m-max",
                    "tags": {
                        "host": ".*",
                    },
                    "filters" : [
                        {
                           "type" : "wildcard",
                           "tagk" : "host",
                           "filter" : ".*",
                           "groupBy" : True
                        },
                        {
                           "type" : "literal_or",
                           "tagk" : "rack",
                           "filter" : "r1|r2",
                           "groupBy" : False
                        }
                    ]
                }
            ]
        }

        await websocket.send(json.dumps(query))
        print('WEBSOCKET OUT -> {}'.format(query))
        version_resp = await websocket.recv()
        print('WEBSOCKET RET <- {}'.format(version_resp))
        return version_resp


loop = asyncio.get_event_loop()
result = loop.run_until_complete(connect_timely())