<s> import os <EOL> import struct <EOL> from ws4py . framing import Frame , OPCODE_CONTINUATION , OPCODE_TEXT , OPCODE_BINARY , OPCODE_CLOSE , OPCODE_PING , OPCODE_PONG <EOL> from ws4py . compat import unicode , py3k <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> class Message ( object ) : <EOL> def __init__ ( self , opcode , data = b'<STR_LIT>' , encoding = '<STR_LIT:utf-8>' ) : <EOL> """<STR_LIT>""" <EOL> self . opcode = opcode <EOL> self . _completed = False <EOL> self . encoding = encoding <EOL> if isinstance ( data , unicode ) : <EOL> if not encoding : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> data = data . encode ( encoding ) <EOL> elif isinstance ( data , bytearray ) : <EOL> data = bytes ( data ) <EOL> elif not isinstance ( data , bytes ) : <EOL> raise TypeError ( "<STR_LIT>" % type ( data ) ) <EOL> self . data = data <EOL> def single ( self , mask = False ) : <EOL> """<STR_LIT>""" <EOL> mask = os . urandom ( <NUM_LIT:4> ) if mask else None <EOL> return Frame ( body = self . data , opcode = self . opcode , <EOL> masking_key = mask , fin = <NUM_LIT:1> ) . build ( ) <EOL> def fragment ( self , first = False , last = False , mask = False ) : <EOL> """<STR_LIT>""" <EOL> fin = <NUM_LIT:1> if last is True else <NUM_LIT:0> <EOL> opcode = self . opcode if first is True else OPCODE_CONTINUATION <EOL> mask = os . urandom ( <NUM_LIT:4> ) if mask else None <EOL> return Frame ( body = self . data , <EOL> opcode = opcode , masking_key = mask , <EOL> fin = fin ) . build ( ) <EOL> @ property <EOL> def completed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _completed <EOL> @ completed . setter <EOL> def completed ( self , state ) : <EOL> """<STR_LIT>""" <EOL> self . _completed = state <EOL> def extend ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( data , bytes ) : <EOL> self . data += data <EOL> elif isinstance ( data , bytearray ) : <EOL> self . data += bytes ( data ) <EOL> elif isinstance ( data , unicode ) : <EOL> self . data += data . encode ( self . encoding ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" % type ( data ) ) <EOL> def __len__ ( self ) : <EOL> return len ( self . __unicode__ ( ) ) <EOL> def __str__ ( self ) : <EOL> if py3k : <EOL> return self . data . decode ( self . encoding ) <EOL> return self . data <EOL> def __unicode__ ( self ) : <EOL> return self . data . decode ( self . encoding ) <EOL> class TextMessage ( Message ) : <EOL> def __init__ ( self , text = None ) : <EOL> Message . __init__ ( self , OPCODE_TEXT , text ) <EOL> @ property <EOL> def is_binary ( self ) : <EOL> return False <EOL> @ property <EOL> def is_text ( self ) : <EOL> return True <EOL> class BinaryMessage ( Message ) : <EOL> def __init__ ( self , bytes = None ) : <EOL> Message . __init__ ( self , OPCODE_BINARY , bytes , encoding = None ) <EOL> @ property <EOL> def is_binary ( self ) : <EOL> return True <EOL> @ property <EOL> def is_text ( self ) : <EOL> return False <EOL> def __len__ ( self ) : <EOL> return len ( self . data ) <EOL> class CloseControlMessage ( Message ) : <EOL> def __init__ ( self , code = <NUM_LIT:1000> , reason = '<STR_LIT>' ) : <EOL> data = b"<STR_LIT>" <EOL> if code : <EOL> data += struct . pack ( "<STR_LIT>" , code ) <EOL> if reason is not None : <EOL> if isinstance ( reason , unicode ) : <EOL> reason = reason . encode ( '<STR_LIT:utf-8>' ) <EOL> data += reason <EOL> Message . __init__ ( self , OPCODE_CLOSE , data , '<STR_LIT:utf-8>' ) <EOL> self . code = code <EOL> self . reason = reason <EOL> def __str__ ( self ) : <EOL> if py3k : <EOL> return self . reason . decode ( '<STR_LIT:utf-8>' ) <EOL> return self . reason <EOL> def __unicode__ ( self ) : <EOL> return self . reason . decode ( self . encoding ) <EOL> class PingControlMessage ( Message ) : <EOL> def __init__ ( self , data = None ) : <EOL> Message . __init__ ( self , OPCODE_PING , data ) <EOL> class PongControlMessage ( Message ) : <EOL> def __init__ ( self , data ) : <EOL> Message . __init__ ( self , OPCODE_PONG , data ) </s>
