Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteBuf> { @Override protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> out) throws Exception { LOGGER.trace("(-->IN): {}, {}, {}", channelHandlerContext, byteBuf, out); if (byteBuf.readableBytes() < MagicCookie.NUM_BYTES + TransmitterAddress.NUM_BYTES + ReceiverAddress.NUM_BYTES + FragmentNumber.NUM_BYTES) { return; } MagicCookie magicCookie = MagicCookie.of(byteBuf); TransmitterAddress transmitterAddress = TransmitterAddress.of(byteBuf); ReceiverAddress receiverAddress = ReceiverAddress.of(byteBuf); FragmentNumber fragmentNumber = FragmentNumber.of(byteBuf); int expectedFrameNumber = fragmentCounter.get() - 1; if (expectedFrameNumber < 0) { expectedFrameNumber = 255; } if (fragmentNumber.getAsByte() != expectedFrameNumber) { LOGGER.warn("Unexpected fragment {} received. Expected {}", fragmentNumber, expectedFrameNumber); } UserDataLength userDataLength = UserDataLength.of(byteBuf); UserData userData; byte userDataLengthAsByte = userDataLength.getAsByte(); if (byteBuf.readableBytes() < userDataLengthAsByte) { return; } if (userDataLengthAsByte > 0) { byte[] userDataByteArray = new byte[userDataLengthAsByte]; byteBuf.readBytes(userDataByteArray); userData = UserData.of(userDataByteArray); } else { userData = UserData.EMPTY; } CRC crc = CRC.of(byteBuf); ScheduledFuture<?> scheduledFuture = currentRetryer.get(); if (scheduledFuture != null) { scheduledFuture.cancel(false); } Runnable postAction = null; switch (magicCookie.getAsInt()) { case AmsSerialFrame.ID: AmsSerialFrame amsSerialFrame = AmsSerialFrame.of(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, userData, crc); LOGGER.debug("Ams Serial Frame received {}", amsSerialFrame); postAction = () -> { ChannelFuture channelFuture = channelHandlerContext.writeAndFlush(AmsSerialAcknowledgeFrame.of(transmitterAddress, receiverAddress, fragmentNumber).getByteBuf()); try { channelFuture.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new PlcRuntimeException(e); } out.add(userData.getByteBuf()); }; break; case AmsSerialAcknowledgeFrame.ID: AmsSerialAcknowledgeFrame amsSerialAcknowledgeFrame = AmsSerialAcknowledgeFrame.of(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, crc); LOGGER.debug("Ams Serial ACK Frame received {}", amsSerialAcknowledgeFrame); ReferenceCountUtil.release(byteBuf); break; case AmsSerialResetFrame.ID: AmsSerialResetFrame amsSerialResetFrame = AmsSerialResetFrame.of(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, crc); LOGGER.debug("Ams Serial Reset Frame received {}", amsSerialResetFrame); ReferenceCountUtil.release(byteBuf); break; default: throw new PlcProtocolException("Unknown type: " + magicCookie); } CRC calculatedCrc = CRC.of(DigestUtil.calculateCrc16(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, userData)); if (!crc.equals(calculatedCrc)) { throw new PlcProtocolException("CRC checksum wrong. Got " + crc + " expected " + calculatedCrc); } if (postAction != null) { postAction.run(); } if (byteBuf.readableBytes() > 0) { throw new IllegalStateException("Unread bytes left: " + byteBuf.readableBytes()); } }    }
@Test public void decode() throws Exception { ArrayList<Object> out = new ArrayList<>(); SUT.decode(channelHandlerContextMock, amsSerialFrame.getByteBuf(), out); assertThat(out, hasSize(1)); }