diff --git a/ChangeLog.txt b/ChangeLog.txt index a8a3f9ba..a9c05696 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,3 +1,7 @@ +Version 0.9: + Moved canRead check to topic from publish phase to subscribe (#144) + Added javadoc and src jars creations for JCenter/Maven central + Version 0.8: Introduced authenticator and authorizator implementations. Fixed bug, doesn't close the channel when connect doesn't contains a valid proto name, issue #65 diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 93550cec..54575880 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -659,16 +659,16 @@ public class ProtocolProcessor { String user = NettyUtils.userName(channel); List newSubscriptions = new ArrayList<>(); for (SubscribeMessage.Couple req : msg.subscriptions()) { - if (!m_authorizator.canRead(req.getTopicFilter(), user, clientSession.clientID)) { + if (!m_authorizator.canRead(req.topicFilter, user, clientSession.clientID)) { //send SUBACK with 0x80, the user hasn't credentials to read the topic - LOG.debug("topic {} doesn't have read credentials", req.getTopicFilter()); + LOG.debug("topic {} doesn't have read credentials", req.topicFilter); ackMessage.addType( AbstractMessage.QOSType.FAILURE); continue; } - AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos()); - Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos); - boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription); + AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.qos); + Subscription newSubscription = new Subscription(clientID, req.topicFilter, qos); + boolean valid = clientSession.subscribe(req.topicFilter, newSubscription); ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE); if (valid) { newSubscriptions.add(newSubscription); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 693bbbd3..db605690 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -73,6 +73,8 @@ public class ProtocolProcessorTest { connMsg.setProtocolVersion((byte) 0x03); m_channel = new EmbeddedChannel(); + NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); + NettyUtils.cleanSession(m_channel, false); //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); @@ -211,8 +213,6 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_channel, msg); @@ -235,8 +235,6 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_channel, msg); @@ -252,8 +250,6 @@ public class ProtocolProcessorTest { public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -273,8 +269,6 @@ public class ProtocolProcessorTest { public void testSubscribeWithBadFormattedTopic() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); - NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_channel, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -298,8 +292,6 @@ public class ProtocolProcessorTest { UnsubscribeMessage msg = new UnsubscribeMessage(); msg.setMessageID(1); msg.addTopicFilter(BAD_FORMATTED_TOPIC); - NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_channel, false); //Exercise m_processor.processUnsubscribe(m_channel, msg); diff --git a/netty_parser/src/main/java/io/moquette/parser/netty/SubscribeEncoder.java b/netty_parser/src/main/java/io/moquette/parser/netty/SubscribeEncoder.java index 55f16bef..e55269d0 100644 --- a/netty_parser/src/main/java/io/moquette/parser/netty/SubscribeEncoder.java +++ b/netty_parser/src/main/java/io/moquette/parser/netty/SubscribeEncoder.java @@ -41,8 +41,8 @@ class SubscribeEncoder extends DemuxEncoder { try { variableHeaderBuff.writeShort(message.getMessageID()); for (SubscribeMessage.Couple c : message.subscriptions()) { - variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopicFilter())); - variableHeaderBuff.writeByte(c.getQos()); + variableHeaderBuff.writeBytes(Utils.encodeString(c.topicFilter)); + variableHeaderBuff.writeByte(c.qos); } int variableHeaderSize = variableHeaderBuff.readableBytes(); diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeDecoderTest.java b/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeDecoderTest.java index bcf3b04d..3b5a932b 100644 --- a/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeDecoderTest.java +++ b/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeDecoderTest.java @@ -119,8 +119,8 @@ public class SubscribeDecoderTest { ByteBuf topicBuffer = Unpooled.buffer(4); topicBuffer.writeShort(messageID); for (SubscribeMessage.Couple couple : topics) { - topicBuffer.writeBytes(Utils.encodeString(couple.getTopicFilter())); - topicBuffer.writeByte(couple.getQos()); + topicBuffer.writeBytes(Utils.encodeString(couple.topicFilter)); + topicBuffer.writeByte(couple.qos); } buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02) @@ -131,8 +131,8 @@ public class SubscribeDecoderTest { private void initPollutedTopic(ByteBuf buff, int messageID, SubscribeMessage.Couple topic) throws IllegalAccessException { ByteBuf topicBuffer = Unpooled.buffer(4); topicBuffer.writeShort(messageID); - topicBuffer.writeBytes(Utils.encodeString(topic.getTopicFilter())); - topicBuffer.writeByte(0xF0 | topic.getQos()); + topicBuffer.writeBytes(Utils.encodeString(topic.topicFilter)); + topicBuffer.writeByte(0xF0 | topic.qos); buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02) .writeBytes(Utils.encodeRemainingLength(topicBuffer.readableBytes())); diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeEncoderTest.java b/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeEncoderTest.java index 11ceaf7c..1578c490 100644 --- a/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeEncoderTest.java +++ b/netty_parser/src/test/java/io/moquette/parser/netty/SubscribeEncoderTest.java @@ -64,10 +64,10 @@ public class SubscribeEncoderTest { assertEquals((byte)0xBB, m_out.readByte()); //Variable part - TestUtils.verifyString(c1.getTopicFilter(), m_out); - assertEquals(c1.getQos(), m_out.readByte()); - TestUtils.verifyString(c2.getTopicFilter(), m_out); - assertEquals(c2.getQos(), m_out.readByte()); + TestUtils.verifyString(c1.topicFilter, m_out); + assertEquals(c1.qos, m_out.readByte()); + TestUtils.verifyString(c2.topicFilter, m_out); + assertEquals(c2.qos, m_out.readByte()); } @Test(expected = IllegalArgumentException.class) diff --git a/parser_commons/src/main/java/io/moquette/proto/messages/SubscribeMessage.java b/parser_commons/src/main/java/io/moquette/proto/messages/SubscribeMessage.java index eb54c114..48776286 100644 --- a/parser_commons/src/main/java/io/moquette/proto/messages/SubscribeMessage.java +++ b/parser_commons/src/main/java/io/moquette/proto/messages/SubscribeMessage.java @@ -26,21 +26,14 @@ public class SubscribeMessage extends MessageIDMessage { public static class Couple { - private byte m_qos; - private String m_topicFilter; + public final byte qos; + public final String topicFilter; public Couple(byte qos, String topic) { - m_qos = qos; - m_topicFilter = topic; + this.qos = qos; + this.topicFilter = topic; } - public byte getQos() { - return m_qos; - } - - public String getTopicFilter() { - return m_topicFilter; - } } private List m_subscriptions = new ArrayList<>();