#NEWLINE# def received_message( self , m ) : #NEWLINE# #INDENT# logger . debug ( ( '%s: Got message from server of type %s' % ( self . request_id , str ( type ( m ) ) ) ) ) #NEWLINE# if ( self . state == self . __class__ . STATE_CONNECTED ) : #NEWLINE# #INDENT# props = json . loads ( str ( m ) ) #NEWLINE# content_type = props [ 'content_type' ] #NEWLINE# self . request_id = props [ 'id' ] #NEWLINE# self . num_segments = 0 #NEWLINE# self . decoder_pipeline . init_request ( self . request_id , content_type ) #NEWLINE# self . last_decoder_message = time . time ( ) #NEWLINE# thread . start_new_thread ( self . guard_timeout , ( ) ) #NEWLINE# logger . info ( ( '%s: Started timeout guard' % self . request_id ) ) #NEWLINE# logger . info ( ( '%s: Initialized request' % self . request_id ) ) #NEWLINE# self . state = self . STATE_INITIALIZED #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( m . data == 'EOS' ) : #NEWLINE# #INDENT# if ( ( self . state != m . STATE_CANCELLING ) and ( self . state != self . STATE_EOS_RECEIVED ) and ( self . state != self . STATE_FINISHED ) ) : #NEWLINE# #INDENT# self . decoder_pipeline . end_request ( ) #NEWLINE# self . state = self . STATE_EOS_RECEIVED #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# logger . info ( ( '%s: Ignoring EOS, worker already in state %d' % ( self . request_id , self . state ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( ( self . state != self . STATE_CANCELLING ) and ( self . state != self . STATE_EOS_RECEIVED ) and ( self . state != self . STATE_FINISHED ) ) : #NEWLINE# #INDENT# if isinstance ( m , ws4py . messaging . BinaryMessage ) : #NEWLINE# #INDENT# self . decoder_pipeline . process_data ( m . data ) #NEWLINE# self . state = self . STATE_PROCESSING #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( m , ws4py . messaging . TextMessage ) : #NEWLINE# #INDENT# props = json . loads ( str ( m ) ) #NEWLINE# if ( 'adaptation_state' in props ) : #NEWLINE# #INDENT# as_props = props [ 'adaptation_state' ] #NEWLINE# if ( as_props . get ( 'type' , '' ) == 'string+gzip+base64' ) : #NEWLINE# #INDENT# adaptation_state = zlib . decompress ( base64 . b64decode ( as_props . get ( 'value' , '' ) ) ) #NEWLINE# logger . info ( ( '%s: Setting adaptation state to user-provided value' % self . request_id ) ) #NEWLINE# self . decoder_pipeline . set_adaptation_state ( adaptation_state ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# logger . warning ( ( '%s: Cannot handle adaptation state type ' % ( self . request_id , as_props . get ( 'type' , '' ) ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# logger . warning ( ( "%s: Got JSON message but don't know what to do with it" % self . request_id ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# logger . info ( ( '%s: Ignoring data, worker already in state %d' % ( self . request_id , self . state ) ) )