wsstat package

Submodules

wsstat.clients module

class wsstat.clients.ConnectedWebsocketConnection(ws, identifier)

Bases: object

increment_message_counter()

message_count

process_message(message)

class wsstat.clients.WebsocketTestingClient(websocket_url, **kwargs)

Bases: object

Setting up the websocket calls the following callbacks that can be overridden to extend functinality. For an example see WebsocketTestingClientWithApiTokenHeader

def before_connect(self): def setup_websocket_connection(self, statedict): def get_identifier(self, statedict): def after_connect(self, statedict):

def before_recv(self, statedict): def after_recv(self, statedict, message):

after_connect(statedict)

after_recv(statedict, message)

before_connect()

before_recv(statedict)

create_websocket_connection()

exit()

get_identifier(statedict)

handle_exceptions(loop, context)

handle_keypresses(keypress)

log(identifier, message)

messages_per_second

setup_tasks()

setup_websocket_connection(statedict)

update_urwid()

class wsstat.clients.WebsocketTestingClientWithRandomApiTokenHeader(*args, **kwargs)

Bases: wsstat.clients.WebsocketTestingClient

Introduces a new parameter: header_name - used to specify the key to ‘extra_headers’ passed to websocket.connect

before_connect()

get_identifier(statedict)

setup_websocket_connection(statedict)

wsstat.gui module

class wsstat.gui.BlinkBoardWidget

Bases: object

generate_blinkers(connected_sockets)

get_ws_status(websocket, connected, error)

update_large_blinkers(blinkers)

update_small_blinkers(blinkers)

class wsstat.gui.LoggerWidget

Bases: object

graph_title = ' Messages / Second | 3 Second window '

log(string)

log_messages = []

timelogger = deque([], maxlen=30)

update_graph_data(data)

class wsstat.gui.WSStatConsoleApplication(client)

Bases: object

class DummyScreen(input=<_io.TextIOWrapper name='<stdin>' mode='r' encoding='UTF-8'>, output=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>)

Bases: urwid.raw_display.Screen

draw_screen(xxx_todo_changeme, r)

class FixedAsyncLoop(**kwargs)

Bases: urwid.main_loop.AsyncioEventLoop

run()

Start the event loop. Exit the loop when any callback raises an exception. If ExitMainLoop is raised, exit cleanly.

run()

wsstat.main module

wsstat.main.parse_args()

wsstat.main.wsstat_console()

Module contents