<s> from __future__ import absolute_import <EOL> import json <EOL> import six <EOL> __all__ = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> class ConnectionRequest ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:host>' , <EOL> '<STR_LIT:path>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:version>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ) <EOL> def __init__ ( self , peer , headers , host , path , params , version , origin , protocols , extensions ) : <EOL> """<STR_LIT>""" <EOL> self . peer = peer <EOL> self . headers = headers <EOL> self . host = host <EOL> self . path = path <EOL> self . params = params <EOL> self . version = version <EOL> self . origin = origin <EOL> self . protocols = protocols <EOL> self . extensions = extensions <EOL> def __json__ ( self ) : <EOL> return { '<STR_LIT>' : self . peer , <EOL> '<STR_LIT>' : self . headers , <EOL> '<STR_LIT:host>' : self . host , <EOL> '<STR_LIT:path>' : self . path , <EOL> '<STR_LIT>' : self . params , <EOL> '<STR_LIT:version>' : self . version , <EOL> '<STR_LIT>' : self . origin , <EOL> '<STR_LIT>' : self . protocols , <EOL> '<STR_LIT>' : self . extensions } <EOL> def __str__ ( self ) : <EOL> return json . dumps ( self . __json__ ( ) ) <EOL> class ConnectionResponse ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:version>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ) <EOL> def __init__ ( self , peer , headers , version , protocol , extensions ) : <EOL> """<STR_LIT>""" <EOL> self . peer = peer <EOL> self . headers = headers <EOL> self . version = version <EOL> self . protocol = protocol <EOL> self . extensions = extensions <EOL> def __json__ ( self ) : <EOL> return { '<STR_LIT>' : self . peer , <EOL> '<STR_LIT>' : self . headers , <EOL> '<STR_LIT:version>' : self . version , <EOL> '<STR_LIT>' : self . protocol , <EOL> '<STR_LIT>' : self . extensions } <EOL> def __str__ ( self ) : <EOL> return json . dumps ( self . __json__ ( ) ) <EOL> class ConnectionAccept ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , subprotocol = None , headers = None ) : <EOL> """<STR_LIT>""" <EOL> assert ( subprotocol is None or type ( subprotocol ) == six . text_type ) <EOL> assert ( headers is None or type ( headers ) == dict ) <EOL> if headers is not None : <EOL> for k , v in headers . items ( ) : <EOL> assert ( type ( k ) == six . text_type ) <EOL> assert ( type ( v ) == six . text_type or type ( v ) == list or type ( v ) == tuple ) <EOL> if type ( v ) == list or type ( v ) == tuple : <EOL> for vv in v : <EOL> assert ( type ( vv ) == six . text_type ) <EOL> self . subprotocol = subprotocol <EOL> self . headers = headers <EOL> class ConnectionDeny ( Exception ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT:code>' , '<STR_LIT>' ) <EOL> BAD_REQUEST = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> FORBIDDEN = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> NOT_FOUND = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> NOT_ACCEPTABLE = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> REQUEST_TIMEOUT = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> INTERNAL_SERVER_ERROR = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> NOT_IMPLEMENTED = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> SERVICE_UNAVAILABLE = <NUM_LIT> <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , code , reason = None ) : <EOL> """<STR_LIT>""" <EOL> assert ( type ( code ) == int ) <EOL> assert ( reason is None or type ( reason ) == six . text_type ) <EOL> self . code = code <EOL> self . reason = reason <EOL> class Message ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> class IncomingMessage ( Message ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , payload , is_binary = False ) : <EOL> """<STR_LIT>""" <EOL> assert ( type ( payload ) == bytes ) <EOL> assert ( type ( is_binary ) == bool ) <EOL> self . payload = payload <EOL> self . is_binary = is_binary <EOL> class OutgoingMessage ( Message ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , payload , is_binary = False , dont_compress = False ) : <EOL> """<STR_LIT>""" <EOL> assert ( type ( payload ) == bytes ) <EOL> assert ( type ( is_binary ) == bool ) <EOL> assert ( type ( dont_compress ) == bool ) <EOL> self . payload = payload <EOL> self . is_binary = is_binary <EOL> self . dont_compress = dont_compress <EOL> class Ping ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' ) <EOL> def __init__ ( self , payload = None ) : <EOL> """<STR_LIT>""" <EOL> assert ( payload is None or type ( payload ) == bytes ) , ( "<STR_LIT>" . format ( type ( payload ) ) ) <EOL> if payload is not None : <EOL> assert ( len ( payload ) < <NUM_LIT> ) , ( "<STR_LIT>" . format ( len ( payload ) ) ) <EOL> self . payload = payload </s>
