@ Override public void messageReceived ( NextFilter nextFilter , IoSession session , Object message ) throws Exception { LOGGER . debug ( "Processing a MESSAGE_RECEIVED for session {}" , session . getId ( ) ) ; if ( ! ( message instanceof IoBuffer ) ) { nextFilter . messageReceived ( session , message ) ; return ; } IoBuffer in = ( IoBuffer ) message ; ProtocolDecoder decoder = getDecoder ( session ) ; ProtocolDecoderOutput decoderOut = getDecoderOut ( session , nextFilter ) ; IoSessionEx sessionEx = ( IoSessionEx ) session ; Thread ioThread = sessionEx . getIoThread ( ) ; // Loop until we don't have anymore byte in the buffer, // or until the decoder throws an unrecoverable exception or // can't decoder a message, because there are not enough // data in the buffer // Note: break out to let old I/O thread unwind after deregistering while ( in . hasRemaining ( ) ) { // If the session is realigned, let the new thread deal with the decoding if ( sessionEx . getIoThread ( ) != ioThread ) { if ( LOGGER . isTraceEnabled ( ) ) { LOGGER . trace ( format ( "Decoding for session=%s will be continued by new thread=%s old thread=%s" , session , sessionEx . getIoThread ( ) , ioThread ) ) ; } break ; } int oldPos = in . position ( ) ; try { synchronized ( decoderOut ) { // Call the decoder with the read bytes decoder . decode ( session , in , decoderOut ) ; // Finish decoding if no exception was thrown. decoderOut . flush ( nextFilter , session ) ; } } catch ( Throwable t ) { ProtocolDecoderException pde ; if ( t instanceof ProtocolDecoderException ) { pde = ( ProtocolDecoderException ) t ; } else { pde = new ProtocolDecoderException ( t ) ; } if ( pde . getHexdump ( ) == null ) { // Generate a message hex dump int curPos = in . position ( ) ; in . position ( oldPos ) ; pde . setHexdump ( in . getHexDump ( ) ) ; in . position ( curPos ) ; } // Fire the exceptionCaught event. synchronized ( decoderOut ) { decoderOut . flush ( nextFilter , session ) ; } nextFilter . exceptionCaught ( session , pde ) ; // Retry only if the type of the caught exception is // recoverable and the buffer position has changed. // We check buffer position additionally to prevent an // infinite loop. if ( ! ( t instanceof RecoverableProtocolDecoderException ) || ( in . position ( ) == oldPos ) ) { break ; } } } }
Process the incoming message calling the session decoder . As the incoming buffer might contains more than one messages we have to loop until the decoder throws an exception .