@Test public void testHeaderWithMessageID() throws Exception { m_buff = Unpooled.buffer(14); int messageID = 123; initHeaderWithMessageID(m_buff, messageID); m_msgdec.decode(null, m_buff, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); assertNotNull(message); assertEquals("Fake Topic", message.getTopicName()); assertEquals(messageID, (int) message.getMessageID()); }
@Override void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { LOG.info("decode invoked with buffer " + in); in.resetReaderIndex(); int startPos = in.readerIndex(); PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { LOG.info("decode ask for more data after " + in); in.resetReaderIndex(); return; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { in.resetReaderIndex(); return; } message.setTopicName(topic); if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) { message.setMessageID(in.readUnsignedShort()); } int stopPos = in.readerIndex(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.readableBytes() < payloadSize) { in.resetReaderIndex(); return; } byte[] b = new byte[payloadSize]; in.readBytes(b); message.setPayload(b); out.add(message); }
PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { LOG.info("decode invoked with buffer " + in); in.resetReaderIndex(); int startPos = in.readerIndex(); PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { LOG.info("decode ask for more data after " + in); in.resetReaderIndex(); return; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { in.resetReaderIndex(); return; } message.setTopicName(topic); if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) { message.setMessageID(in.readUnsignedShort()); } int stopPos = in.readerIndex(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.readableBytes() < payloadSize) { in.resetReaderIndex(); return; } byte[] b = new byte[payloadSize]; in.readBytes(b); message.setPayload(b); out.add(message); } }
PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { LOG.info("decode invoked with buffer " + in); in.resetReaderIndex(); int startPos = in.readerIndex(); PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { LOG.info("decode ask for more data after " + in); in.resetReaderIndex(); return; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { in.resetReaderIndex(); return; } message.setTopicName(topic); if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) { message.setMessageID(in.readUnsignedShort()); } int stopPos = in.readerIndex(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.readableBytes() < payloadSize) { in.resetReaderIndex(); return; } byte[] b = new byte[payloadSize]; in.readBytes(b); message.setPayload(b); out.add(message); }  }
PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { LOG.info("decode invoked with buffer " + in); in.resetReaderIndex(); int startPos = in.readerIndex(); PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { LOG.info("decode ask for more data after " + in); in.resetReaderIndex(); return; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { in.resetReaderIndex(); return; } message.setTopicName(topic); if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) { message.setMessageID(in.readUnsignedShort()); } int stopPos = in.readerIndex(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.readableBytes() < payloadSize) { in.resetReaderIndex(); return; } byte[] b = new byte[payloadSize]; in.readBytes(b); message.setPayload(b); out.add(message); }   }
PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { LOG.info("decode invoked with buffer " + in); in.resetReaderIndex(); int startPos = in.readerIndex(); PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { LOG.info("decode ask for more data after " + in); in.resetReaderIndex(); return; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { in.resetReaderIndex(); return; } message.setTopicName(topic); if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) { message.setMessageID(in.readUnsignedShort()); } int stopPos = in.readerIndex(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.readableBytes() < payloadSize) { in.resetReaderIndex(); return; } byte[] b = new byte[payloadSize]; in.readBytes(b); message.setPayload(b); out.add(message); }    }