client.py 文件源码

def _ws_events(ws_conn, message, snapshot, since, on_message, on_error):
    """Process websocket events."""
    # Pylint complains too many nested blocks.
    #
    # pylint: disable=R0101
    last_timestamp = since
    subscription_msg = {'since': since,
                        'snapshot': snapshot}
    subscription_msg.update(message)

    try:
        ws_conn.send(json.dumps(subscription_msg))
        while True:
            try:
                reply = ws_conn.recv()
                if not reply:
                    break

                result = json.loads(reply)
                if '_error' in result:
                    if on_error:
                        on_error(result)
                    break

                last_timestamp = result.get('when', time.time())
                if on_message:
                    if not on_message(result):
                        break
            except ws_client.WebSocketTimeoutException:
                ws_conn.ping()

    except ws_client.WebSocketConnectionClosedException as err:
        _LOGGER.debug('ws connection closed, will retry: %s.', str(err))
        raise _RetryError(last_timestamp)
    finally:
        ws_conn.close()