Payload2TcpProtocol extends MessageToMessageCodec<ByteBuf, ByteBuf> { @SuppressWarnings("unchecked") @Override protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> out) { if (byteBuf == Unpooled.EMPTY_BUFFER) { return; } LOGGER.trace("(-->IN): {}, {}, {}", channelHandlerContext, byteBuf, out); retainingBuf = Unpooled.wrappedBuffer(retainingBuf, byteBuf.retain()); while (retainingBuf.readableBytes() >= AmsTcpHeader.Reserved.NUM_BYTES + TcpLength.NUM_BYTES && retainingBuf.readableBytes() >= retainingBuf.getUnsignedIntLE(retainingBuf.readerIndex() + AmsTcpHeader.Reserved.NUM_BYTES) + AmsTcpHeader.Reserved.NUM_BYTES + TcpLength.NUM_BYTES) { retainingBuf.skipBytes(AmsTcpHeader.Reserved.NUM_BYTES); TcpLength packetLength = TcpLength.of(retainingBuf); AmsTcpHeader amsTcpHeader = AmsTcpHeader.of(packetLength); LOGGER.debug("AMS TCP Header {}", amsTcpHeader); out.add(retainingBuf.readBytes((int) packetLength.getAsLong())); } if (retainingBuf.readableBytes() == 0) { retainingBuf.release(); retainingBuf = Unpooled.EMPTY_BUFFER; } }    }
@Test public void decode() throws Exception { ArrayList<Object> out = new ArrayList<>(); SUT.decode(channelHandlerContextMock, amsTCPPacket.getByteBuf(), out); assertThat(out, hasSize(1)); }
@Test public void fragmentedDecode() throws Exception { ArrayList<Object> out = new ArrayList<>(); ByteBuf byteBuf = amsTCPPacket.getByteBuf(); SUT.decode(channelHandlerContextMock, byteBuf.readBytes(byteBuf.readableBytes() - 2), out); assertThat(out, hasSize(0)); SUT.decode(channelHandlerContextMock, byteBuf, out); assertThat(out, hasSize(1)); byteBuf.release(); }
@Test public void multipleMessagesDecode() throws Exception { ArrayList<Object> out = new ArrayList<>(); ByteBuf byteBuf = Unpooled.wrappedBuffer(amsTCPPacket.getByteBuf(), amsTCPPacket.getByteBuf()); SUT.decode(channelHandlerContextMock, byteBuf, out); assertThat(out, hasSize(2)); byteBuf.release(); }