Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteBuf> { @Override protected void encode(ChannelHandlerContext channelHandlerContext, ByteBuf amsPacket, List<Object> out) throws PlcProtocolPayloadTooBigException { if (amsPacket == Unpooled.EMPTY_BUFFER) { ScheduledFuture<?> scheduledFuture = currentRetryer.get(); if (scheduledFuture != null) { scheduledFuture.cancel(true); } return; } LOGGER.trace("(<--OUT): {}, {}, {}", channelHandlerContext, amsPacket, out); int fragmentNumber = fragmentCounter.getAndUpdate(value -> value > 255 ? 0 : ++value); LOGGER.debug("Using fragmentNumber {} for {}", fragmentNumber, amsPacket); UserData userData = UserData.of(amsPacket); if (userData.getCalculatedLength() > 255) { throw new PlcProtocolPayloadTooBigException("ADS/AMS", 255, (int) userData.getCalculatedLength(), amsPacket); } AmsSerialFrame amsSerialFrame = AmsSerialFrame.of(FragmentNumber.of((byte) fragmentNumber), userData); MutableInt retryCount = new MutableInt(0); ScheduledFuture<?> oldRetryer = currentRetryer.get(); if (oldRetryer != null) { oldRetryer.cancel(false); } currentRetryer.set(channelHandlerContext.executor().scheduleAtFixedRate(() -> { LOGGER.trace("Retrying {} the {} time", amsSerialFrame, retryCount); int currentTry = retryCount.incrementAndGet(); if (currentTry > 10) { channelHandlerContext.writeAndFlush(AmsSerialResetFrame.of(FragmentNumber.of((byte) fragmentNumber))); PlcRuntimeException plcRuntimeException = new PlcRuntimeException("Retry exhausted after " + retryCount + " times"); channelHandlerContext.fireExceptionCaught(plcRuntimeException); throw plcRuntimeException; } else { channelHandlerContext.writeAndFlush(amsSerialFrame); } }, 100, 100, TimeUnit.MILLISECONDS)); out.add(amsSerialFrame.getByteBuf()); }    }
@Test public void encode() throws Exception { ArrayList<Object> out = new ArrayList<>(); SUT.encode(channelHandlerContextMock, Unpooled.wrappedBuffer(amsPacketBytes), out); assertEquals(1, out.size()); assertThat(out, hasSize(1)); }