@Test public void testDecodeBigContent() throws Exception { int size = 129; IoBuffer payload = TestUtils.generateRandomPayload(size); IoBuffer firstPublish = generatePublishQoS0(payload); IoBuffer secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size)); IoBuffer doubleMessageBuf = IoBuffer.allocate(size * 2).setAutoExpand(true); doubleMessageBuf.put(firstPublish).put(secondPublish).flip(); MessageDecoderResult res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder); assertEquals(MessageDecoder.OK, res); PublishMessage pubMsg = m_mockProtoDecoder.getMessage(); assertNotNull(pubMsg); res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder); assertNotNull(m_mockProtoDecoder.getMessage()); assertEquals(MessageDecoder.OK, res); m_buff.flip(); }
public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { LOG.debug("decode invoked with buffer " + in); int startPos = in.position(); PublishMessage message = new PublishMessage(); if (decodeCommonHeader(message, in) == NEED_DATA) { LOG.info("decode ask for more data after " + in); return NEED_DATA; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { return NEED_DATA; } message.setTopicName(topic); if (message.getQos() == QOSType.LEAST_ONE || message.getQos() == QOSType.EXACTLY_ONCE) { message.setMessageID(Utils.readWord(in)); } int stopPos = in.position(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.remaining() < payloadSize) { return NEED_DATA; } byte[] b = new byte[payloadSize]; in.get(b); message.setPayload(b); out.write(message); return OK; }
PublishDecoder extends MqttDecoder { public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { LOG.debug("decode invoked with buffer " + in); int startPos = in.position(); PublishMessage message = new PublishMessage(); if (decodeCommonHeader(message, in) == NEED_DATA) { LOG.info("decode ask for more data after " + in); return NEED_DATA; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { return NEED_DATA; } message.setTopicName(topic); if (message.getQos() == QOSType.LEAST_ONE || message.getQos() == QOSType.EXACTLY_ONCE) { message.setMessageID(Utils.readWord(in)); } int stopPos = in.position(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.remaining() < payloadSize) { return NEED_DATA; } byte[] b = new byte[payloadSize]; in.get(b); message.setPayload(b); out.write(message); return OK; } }
PublishDecoder extends MqttDecoder { public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { LOG.debug("decode invoked with buffer " + in); int startPos = in.position(); PublishMessage message = new PublishMessage(); if (decodeCommonHeader(message, in) == NEED_DATA) { LOG.info("decode ask for more data after " + in); return NEED_DATA; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { return NEED_DATA; } message.setTopicName(topic); if (message.getQos() == QOSType.LEAST_ONE || message.getQos() == QOSType.EXACTLY_ONCE) { message.setMessageID(Utils.readWord(in)); } int stopPos = in.position(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.remaining() < payloadSize) { return NEED_DATA; } byte[] b = new byte[payloadSize]; in.get(b); message.setPayload(b); out.write(message); return OK; }  }
PublishDecoder extends MqttDecoder { public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { LOG.debug("decode invoked with buffer " + in); int startPos = in.position(); PublishMessage message = new PublishMessage(); if (decodeCommonHeader(message, in) == NEED_DATA) { LOG.info("decode ask for more data after " + in); return NEED_DATA; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { return NEED_DATA; } message.setTopicName(topic); if (message.getQos() == QOSType.LEAST_ONE || message.getQos() == QOSType.EXACTLY_ONCE) { message.setMessageID(Utils.readWord(in)); } int stopPos = in.position(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.remaining() < payloadSize) { return NEED_DATA; } byte[] b = new byte[payloadSize]; in.get(b); message.setPayload(b); out.write(message); return OK; }  MessageDecoderResult decodable(IoSession session, IoBuffer in); MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out); }
PublishDecoder extends MqttDecoder { public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { LOG.debug("decode invoked with buffer " + in); int startPos = in.position(); PublishMessage message = new PublishMessage(); if (decodeCommonHeader(message, in) == NEED_DATA) { LOG.info("decode ask for more data after " + in); return NEED_DATA; } int remainingLength = message.getRemainingLength(); String topic = Utils.decodeString(in); if (topic == null) { return NEED_DATA; } message.setTopicName(topic); if (message.getQos() == QOSType.LEAST_ONE || message.getQos() == QOSType.EXACTLY_ONCE) { message.setMessageID(Utils.readWord(in)); } int stopPos = in.position(); int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); if (in.remaining() < payloadSize) { return NEED_DATA; } byte[] b = new byte[payloadSize]; in.get(b); message.setPayload(b); out.write(message); return OK; }  MessageDecoderResult decodable(IoSession session, IoBuffer in); MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out);  }