ProtocolCodecFilter extends AbstractIoFilter { @SuppressWarnings("unchecked") @Override public void messageReceived(IoSession session, Object in, ReadFilterChainController controller) { LOGGER.debug("Processing a MESSAGE_RECEIVED for session {}", session); DECODING_STATE state = getDecodingState(session); MESSAGE msg; try { while (((msg = decoder.decode((ENCODED) in, state)) != null)) { super.messageReceived(session, msg, controller); } } catch (ProtocolDecoderException e) { LOGGER.debug("decoding exception : ", e); throw e; } } ProtocolCodecFilter(ProtocolEncoder<MESSAGE, ENCODED, ENCODING_STATE> encoder,
            ProtocolDecoder<ENCODED, MESSAGE, DECODING_STATE> decoder); @SuppressWarnings("unchecked") @Override void messageReceived(IoSession session, Object in, ReadFilterChainController controller); @SuppressWarnings("unchecked") @Override void messageWriting(IoSession session, WriteRequest message, WriteFilterChainController controller); @Override void sessionOpened(IoSession session); @Override void sessionClosed(IoSession session);  }
@Test public void loop_decode_twice() { IoSession session = mock(IoSession.class); ByteBuffer buff = ByteBuffer.wrap("test".getBytes()); Object decodingState = new Object(); when(session.getAttribute(new AttributeKey<Object>(Object.class, "internal_decoder"))) .thenReturn(decodingState); Object decoded = new Object(); when(decoder.decode(buff, decodingState)).thenReturn(decoded).thenReturn(decoded).thenReturn(null); ReadFilterChainController ctrl = mock(ReadFilterChainController.class); filter.messageReceived(session, buff, ctrl); verify(decoder, times(3)).decode(buff, decodingState); verify(ctrl, times(2)).callReadNextFilter(decoded); verify(session).getAttribute(new AttributeKey<Object>(Object.class, "internal_decoder")); verifyNoMoreInteractions(encoder, decoder, session, ctrl); }