public void handleMessageReceived ( Packet packet ) { if ( log . isTraceEnabled ( ) ) { log . trace ( "handleMessageReceived - {}" , sessionId ) ; } // set the packet expiration time if maxHandlingTimeout is not disabled (set to 0) if ( maxHandlingTimeout > 0 ) { packet . setExpirationTime ( System . currentTimeMillis ( ) + maxHandlingTimeout ) ; } if ( executor != null ) { final byte dataType = packet . getHeader ( ) . getDataType ( ) ; // route these types outside the executor switch ( dataType ) { case Constants . TYPE_PING : case Constants . TYPE_ABORT : case Constants . TYPE_BYTES_READ : case Constants . TYPE_CHUNK_SIZE : case Constants . TYPE_CLIENT_BANDWIDTH : case Constants . TYPE_SERVER_BANDWIDTH : // pass message to the handler try { handler . messageReceived ( this , packet ) ; } catch ( Exception e ) { log . error ( "Error processing received message {}" , sessionId , e ) ; } break ; default : final String messageType = getMessageType ( packet ) ; try { // increment the packet number final long packetNumber = packetSequence . incrementAndGet ( ) ; if ( executorQueueSizeToDropAudioPackets > 0 && currentQueueSize . get ( ) >= executorQueueSizeToDropAudioPackets ) { if ( packet . getHeader ( ) . getDataType ( ) == Constants . TYPE_AUDIO_DATA ) { // if there's a backlog of messages in the queue. Flash might have sent a burst of messages after a network congestion. Throw away packets that we are able to discard. log . info ( "Queue threshold reached. Discarding packet: session=[{}], msgType=[{}], packetNum=[{}]" , sessionId , messageType , packetNumber ) ; return ; } } int streamId = packet . getHeader ( ) . getStreamId ( ) . intValue ( ) ; if ( log . isTraceEnabled ( ) ) { log . trace ( "Handling message for streamId: {}, channelId: {} Channels: {}" , streamId , packet . getHeader ( ) . getChannelId ( ) , channels ) ; } // create a task to setProcessing the message ReceivedMessageTask task = new ReceivedMessageTask ( sessionId , packet , handler , this ) ; task . setPacketNumber ( packetNumber ) ; // create a task queue ReceivedMessageTaskQueue newStreamTasks = new ReceivedMessageTaskQueue ( streamId , this ) ; // put the queue in the task by stream map ReceivedMessageTaskQueue currentStreamTasks = tasksByStreams . putIfAbsent ( streamId , newStreamTasks ) ; if ( currentStreamTasks != null ) { // add the task to the existing queue currentStreamTasks . addTask ( task ) ; } else { // add the task to the newly created and just added queue newStreamTasks . addTask ( task ) ; } } catch ( Exception e ) { log . error ( "Incoming message handling failed on session=[" + sessionId + "], messageType=[" + messageType + "]" , e ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "Execution rejected on {} - {}" , sessionId , RTMP . states [ getStateCode ( ) ] ) ; log . debug ( "Lock permits - decode: {} encode: {}" , decoderLock . availablePermits ( ) , encoderLock . availablePermits ( ) ) ; } } } } else { log . debug ( "Executor is null on {} state: {}" , sessionId , RTMP . states [ getStateCode ( ) ] ) ; // pass message to the handler try { handler . messageReceived ( this , packet ) ; } catch ( Exception e ) { log . error ( "Error processing received message {} state: {}" , sessionId , RTMP . states [ getStateCode ( ) ] , e ) ; } } }
Handle the incoming message .