diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 9f40542c..499a0c0d 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -224,7 +224,7 @@ class ProtocolProcessor implements EventHandler { //Handle will flag if (msg.isWillFlag()) { - AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; + AbstractMessage.QOSType willQos = AbstractMessage.QOSType.valueOf(msg.getWillQos()); byte[] willPayload = msg.getWillMessage(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); //save the will testament in the clientID store @@ -391,7 +391,7 @@ class ProtocolProcessor implements EventHandler { } for (final Subscription sub : subscriptions.matches(topic)) { AbstractMessage.QOSType qos = publishingQos; - if (qos.ordinal() > sub.getRequestedQos().ordinal()) { + if (qos.byteValue() > sub.getRequestedQos().byteValue()) { qos = sub.getRequestedQos(); } @@ -637,7 +637,7 @@ class ProtocolProcessor implements EventHandler { ackMessage.setMessageID(msg.getMessageID()); for (SubscribeMessage.Couple req : msg.subscriptions()) { - AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; + AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos()); Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession); boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter()); ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index fc4da271..84adac23 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -237,7 +237,7 @@ public class SubscriptionsStore { for (Subscription sub : matchingSubs) { Subscription existingSub = subsForClient.get(sub.getClientId()); //update the selected subscriptions if not present or if has a greater qos - if (existingSub == null || existingSub.getRequestedQos().ordinal() < sub.getRequestedQos().ordinal()) { + if (existingSub == null || existingSub.getRequestedQos().byteValue() < sub.getRequestedQos().byteValue()) { subsForClient.put(sub.getClientId(), sub); } } diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java index fa25a326..e79cd1d5 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -113,7 +113,7 @@ public class ServerLowlevelMessagesIntegrationTests { connectMessage.setWillFlag(true); connectMessage.setWillMessage(willTestamentMsg.getBytes()); connectMessage.setWillTopic(willTestamentTopic); - connectMessage.setWillQos((byte) QOSType.MOST_ONE.ordinal()); + connectMessage.setWillQos(QOSType.MOST_ONE.byteValue()); //Execute m_client.sendMessage(connectMessage); @@ -155,7 +155,7 @@ public class ServerLowlevelMessagesIntegrationTests { connectMessage.setClientID(""); connectMessage.setKeepAlive(keepAlive); connectMessage.setWillFlag(false); - connectMessage.setWillQos((byte) QOSType.MOST_ONE.ordinal()); + connectMessage.setWillQos(QOSType.MOST_ONE.byteValue()); //Execute m_client.sendMessage(connectMessage); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 4c75775b..0efe2c8e 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -199,7 +199,7 @@ public class ProtocolProcessorTest { public void testSubscribe() { //Exercise SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); @@ -213,7 +213,7 @@ public class ProtocolProcessorTest { @Test public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); subscriptions.clearAllSubscriptions(); @@ -234,7 +234,7 @@ public class ProtocolProcessorTest { @Test public void testSubscribeWithBadFormattedTopic() { SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), BAD_FORMATTED_TOPIC)); + msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); subscriptions.clearAllSubscriptions(); @@ -329,7 +329,7 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); - msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); + msg.addSubscription(new SubscribeMessage.Couple(QOSType.MOST_ONE.byteValue(), "#")); m_processor.processSubscribe(m_session, msg/*, FAKE_PUBLISHER_ID, false*/); //Verify diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/DemuxDecoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/DemuxDecoder.java index b54d0255..c742b2b0 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/DemuxDecoder.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/DemuxDecoder.java @@ -73,7 +73,11 @@ abstract class DemuxDecoder { message.setMessageType(messageType); message.setDupFlag(dupFlag); - message.setQos(AbstractMessage.QOSType.values()[qosLevel]); + try { + message.setQos(AbstractMessage.QOSType.valueOf(qosLevel)); + } catch(IllegalArgumentException e) { + throw new CorruptedFrameException(String.format("Received an invalid QOS: %s", e.getMessage()), e); + } message.setRetainFlag(retainFlag); message.setRemainingLength(remainingLength); return true; diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckDecoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckDecoder.java index d4596a40..90848b23 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckDecoder.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckDecoder.java @@ -49,7 +49,7 @@ class SubAckDecoder extends DemuxDecoder { } for (int i = 0; i < remainingLength; i++) { byte qos = in.readByte(); - message.addType(AbstractMessage.QOSType.values()[qos]); + message.addType(AbstractMessage.QOSType.valueOf(qos)); } out.add(message); diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java index a61bea13..e6dff215 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java @@ -40,8 +40,7 @@ class SubAckEncoder extends DemuxEncoder { buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); buff.writeShort(message.getMessageID()); for (QOSType c : message.types()) { - int qosValue = (c == QOSType.FAILURE) ? qosValue = 0x80 : c.ordinal(); - buff.writeByte(qosValue); + buff.writeByte(c.byteValue()); } out.writeBytes(buff); diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java index 16622b2b..7b93d04f 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java @@ -184,7 +184,7 @@ public class Utils { flags |= 0x01; } - flags |= ((message.getQos().ordinal() & 0x03) << 1); + flags |= ((message.getQos().byteValue() & 0x03) << 1); return flags; } diff --git a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckDecoderTest.java b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckDecoderTest.java index 49a2ba9e..2f941182 100644 --- a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckDecoderTest.java +++ b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckDecoderTest.java @@ -81,13 +81,31 @@ public class SubAckDecoderTest { assertEquals(AbstractMessage.QOSType.LEAST_ONE, message.types().get(0)); } + @Test + public void testEncodeWithFailureQOS() throws Exception { + initHeaderQos(m_buff, 0xAABB, AbstractMessage.QOSType.FAILURE); + + //Exercise + m_msgdec.decode(null, m_buff, m_results); + + //Verify + assertFalse(m_results.isEmpty()); + SubAckMessage message = (SubAckMessage)m_results.get(0); + assertNotNull(message); + assertEquals(0xAABB, message.getMessageID().intValue()); + List qoses = message.types(); + assertEquals(1, qoses.size()); + assertEquals(AbstractMessage.QOSType.FAILURE, qoses.get(0)); + assertEquals(AbstractMessage.SUBACK, message.getMessageType()); + } + private void initHeaderQos(ByteBuf buff, int messageID, AbstractMessage.QOSType... qoss) throws IllegalAccessException { buff.clear().writeByte(AbstractMessage.SUBACK << 4). writeBytes(Utils.encodeRemainingLength(2 + qoss.length)); buff.writeShort(messageID); for (AbstractMessage.QOSType qos : qoss) { - buff.writeByte(qos.ordinal()); + buff.writeByte(qos.byteValue()); } } } diff --git a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckEncoderTest.java b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckEncoderTest.java index 797b07ee..caec56a6 100644 --- a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckEncoderTest.java +++ b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckEncoderTest.java @@ -49,7 +49,7 @@ public class SubAckEncoderTest { //Exercise m_encoder.encode(m_mockedContext, msg, m_out); } - + @Test public void testEncodeWithMultipleQos() throws Exception { SubAckMessage msg = new SubAckMessage(); @@ -70,8 +70,29 @@ public class SubAckEncoderTest { //Variable part assertEquals((byte)0xAA, m_out.readByte()); //MessageID MSB assertEquals((byte)0xBB, m_out.readByte()); //MessageID LSB - assertEquals((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), m_out.readByte()); - assertEquals((byte)AbstractMessage.QOSType.LEAST_ONE.ordinal(), m_out.readByte()); - assertEquals((byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), m_out.readByte()); + assertEquals(AbstractMessage.QOSType.MOST_ONE.byteValue(), m_out.readByte()); + assertEquals(AbstractMessage.QOSType.LEAST_ONE.byteValue(), m_out.readByte()); + assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE.byteValue(), m_out.readByte()); + } + + @Test + public void testEncodeWithFailureQOS() throws Exception { + SubAckMessage msg = new SubAckMessage(); + + int messageID = 0xAABB; + msg.setMessageID(messageID); + msg.addType(AbstractMessage.QOSType.FAILURE); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + //Verify + assertEquals((byte) (AbstractMessage.SUBACK << 4 ), m_out.readByte()); //1 byte + assertEquals(3, m_out.readByte()); //remaining length + + //Variable part + assertEquals((byte)0xAA, m_out.readByte()); //MessageID MSB + assertEquals((byte)0xBB, m_out.readByte()); //MessageID LSB + assertEquals((byte)0x80, m_out.readByte()); //Failure return code } } diff --git a/parser_commons/src/main/java/org/eclipse/moquette/proto/Utils.java b/parser_commons/src/main/java/org/eclipse/moquette/proto/Utils.java index 00acc79f..e07dd48a 100644 --- a/parser_commons/src/main/java/org/eclipse/moquette/proto/Utils.java +++ b/parser_commons/src/main/java/org/eclipse/moquette/proto/Utils.java @@ -173,7 +173,7 @@ public class Utils { flags |= 0x01; } - flags |= ((message.getQos().ordinal() & 0x03) << 1); + flags |= ((message.getQos().byteValue() & 0x03) << 1); return flags; } diff --git a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java index b7c40305..e979d887 100644 --- a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java +++ b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java @@ -39,9 +39,39 @@ public abstract class AbstractMessage { public static enum QOSType { MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED, FAILURE; - + + public static QOSType valueOf(byte qos){ + switch(qos) { + case 0x00: + return MOST_ONE; + case 0x01: + return LEAST_ONE; + case 0x02: + return EXACTLY_ONCE; + case (byte) 0x80: + return FAILURE; + default: + throw new IllegalArgumentException("Invalid QOS Type. Expected either 0, 1, 2, or 0x80. Given: " + qos); + } + } + + public byte byteValue() { + switch(this) { + case MOST_ONE: + return 0; + case LEAST_ONE: + return 1; + case EXACTLY_ONCE: + return 2; + case FAILURE: + return (byte) 0x80; + default: + throw new IllegalArgumentException("Cannot give byteValue of QOSType: " + this.name()); + } + } + public static String formatQoS(QOSType qos) { - return String.format("%d - %s", qos.ordinal(), qos.name()); + return String.format("%d - %s", qos.byteValue(), qos.name()); } } //type