public List < Object > decodeBuffer ( RTMPConnection conn , IoBuffer buffer ) { final int position = buffer . position ( ) ; //if (log.isTraceEnabled()) { //log.trace("decodeBuffer: {}", Hex.encodeHexString(Arrays.copyOfRange(buffer.array(), position, buffer.limit()))); //} // decoded results List < Object > result = null ; if ( conn != null ) { //log.trace("Decoding for connection - session id: {}", conn.getSessionId()); try { // instance list to hold results result = new LinkedList <> ( ) ; // get the local decode state RTMPDecodeState state = conn . getDecoderState ( ) ; //if (log.isTraceEnabled()) { //log.trace("RTMP decode state {}", state); //} if ( ! conn . getSessionId ( ) . equals ( state . getSessionId ( ) ) ) { log . warn ( "Session decode overlap: {} != {}" , conn . getSessionId ( ) , state . getSessionId ( ) ) ; } int remaining ; while ( ( remaining = buffer . remaining ( ) ) > 0 ) { if ( state . canStartDecoding ( remaining ) ) { //log.trace("Can start decoding"); state . startDecoding ( ) ; } else { log . trace ( "Cannot start decoding" ) ; break ; } final Object decodedObject = decode ( conn , state , buffer ) ; if ( state . hasDecodedObject ( ) ) { //log.trace("Has decoded object"); if ( decodedObject != null ) { result . add ( decodedObject ) ; } } else if ( state . canContinueDecoding ( ) ) { //log.trace("Can continue decoding"); continue ; } else { log . trace ( "Cannot continue decoding" ) ; break ; } } } catch ( Exception ex ) { log . warn ( "Failed to decodeBuffer: pos {}, limit {}, chunk size {}, buffer {}" , position , buffer . limit ( ) , conn . getState ( ) . getReadChunkSize ( ) , Hex . encodeHexString ( Arrays . copyOfRange ( buffer . array ( ) , position , buffer . limit ( ) ) ) ) ; // catch any non-handshake exception in the decoding; close the connection log . warn ( "Closing connection because decoding failed: {}" , conn , ex ) ; // clear the buffer to eliminate memory leaks when we can't parse protocol buffer . clear ( ) ; // close connection because we can't parse data from it conn . close ( ) ; } finally { //if (log.isTraceEnabled()) { //log.trace("decodeBuffer - post decode input buffer position: {} remaining: {}", buffer.position(), buffer.remaining()); //} buffer . compact ( ) ; } } else { log . error ( "Decoding buffer failed, no current connection!?" ) ; } return result ; }
Decode all available objects in buffer .