From 161421a00c2b1a5db88c3e9c7f4712431c343d53 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sun, 3 Aug 2014 11:17:25 +0200 Subject: [PATCH] Renamed subscribe and unsubscribe topic to topicFilter --- .../moquette/messaging/spi/impl/ProtocolProcessor.java | 4 ++-- .../mqtt/moquette/messaging/spi/impl/SimpleMessaging.java | 2 +- .../dna/mqtt/moquette/parser/netty/SubscribeEncoder.java | 2 +- .../mqtt/moquette/parser/netty/UnsubscribeDecoder.java | 4 ++-- .../mqtt/moquette/parser/netty/UnsubscribeEncoder.java | 4 ++-- .../mqtt/moquette/parser/netty/SubscribeDecoderTest.java | 4 ++-- .../mqtt/moquette/parser/netty/SubscribeEncoderTest.java | 4 ++-- .../moquette/parser/netty/UnsubscribeDecoderTest.java | 5 ++--- .../moquette/parser/netty/UnsubscribeEncoderTest.java | 6 +++--- .../mqtt/moquette/proto/messages/SubscribeMessage.java | 8 ++++---- .../mqtt/moquette/proto/messages/UnsubscribeMessage.java | 4 ++-- 11 files changed, 23 insertions(+), 24 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 1a4c6535..8cdc9a52 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -532,8 +532,8 @@ class ProtocolProcessor implements EventHandler { for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; - Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); - subscribeSingleTopic(newSubscription, req.getTopic()); + Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession); + subscribeSingleTopic(newSubscription, req.getTopicFilter()); } //ack the client diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 26443a26..0dea1bf9 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -169,7 +169,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (message instanceof UnsubscribeMessage) { UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); + m_processor.processUnsubscribe(session, clientID, unsubMsg.topicFilters(), unsubMsg.getMessageID()); } else if (message instanceof SubscribeMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java index 4cb99034..efcec793 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java @@ -41,7 +41,7 @@ class SubscribeEncoder extends DemuxEncoder { try { variableHeaderBuff.writeShort(message.getMessageID()); for (SubscribeMessage.Couple c : message.subscriptions()) { - variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopic())); + variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopicFilter())); variableHeaderBuff.writeByte(c.getQos()); } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java index b8f751b7..bb86f669 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java @@ -48,10 +48,10 @@ class UnsubscribeDecoder extends DemuxDecoder { message.setMessageID(in.readUnsignedShort()); int readed = in.readerIndex()- start; while (readed < message.getRemainingLength()) { - message.addTopic(Utils.decodeString(in)); + message.addTopicFilter(Utils.decodeString(in)); readed = in.readerIndex()- start; } - if (message.topics().isEmpty()) { + if (message.topicFilters().isEmpty()) { throw new CorruptedFrameException("unsubscribe MUST have got at least 1 topic"); } out.add(message); diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java index d29eb3ae..e59e3bd4 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java @@ -29,7 +29,7 @@ class UnsubscribeEncoder extends DemuxEncoder { @Override protected void encode(ChannelHandlerContext chc, UnsubscribeMessage message, ByteBuf out) { - if (message.topics().isEmpty()) { + if (message.topicFilters().isEmpty()) { throw new IllegalArgumentException("Found an unsubscribe message with empty topics"); } @@ -41,7 +41,7 @@ class UnsubscribeEncoder extends DemuxEncoder { ByteBuf buff = null; try { variableHeaderBuff.writeShort(message.getMessageID()); - for (String topic : message.topics()) { + for (String topic : message.topicFilters()) { variableHeaderBuff.writeBytes(Utils.encodeString(topic)); } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java index bd3f2b5a..35b656f8 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java @@ -106,7 +106,7 @@ public class SubscribeDecoderTest { ByteBuf topicBuffer = Unpooled.buffer(4); topicBuffer.writeShort(messageID); for (SubscribeMessage.Couple couple : topics) { - topicBuffer.writeBytes(Utils.encodeString(couple.getTopic())); + topicBuffer.writeBytes(Utils.encodeString(couple.getTopicFilter())); topicBuffer.writeByte(couple.getQos()); } @@ -118,7 +118,7 @@ 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.getTopic())); + topicBuffer.writeBytes(Utils.encodeString(topic.getTopicFilter())); topicBuffer.writeByte(0xF0 | topic.getQos()); buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02) diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java index 36e8d9b7..8be861c8 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java @@ -65,9 +65,9 @@ public class SubscribeEncoderTest { assertEquals((byte)0xBB, m_out.readByte()); //Variable part - verifyString(c1.getTopic(), m_out); + verifyString(c1.getTopicFilter(), m_out); assertEquals(c1.getQos(), m_out.readByte()); - verifyString(c2.getTopic(), m_out); + verifyString(c2.getTopicFilter(), m_out); assertEquals(c2.getQos(), m_out.readByte()); } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java index b5ca9b84..11421e3c 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java @@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.CorruptedFrameException; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; @@ -65,9 +66,7 @@ public class UnsubscribeDecoderTest { assertFalse(m_results.isEmpty()); UnsubscribeMessage message = (UnsubscribeMessage)m_results.get(0); assertNotNull(message); - assertEquals(2, message.topics().size()); - assertEquals(topic1, message.topics().get(0)); - assertEquals(topic2, message.topics().get(1)); + assertEquals(Arrays.asList(topic1, topic2), message.topicFilters()); assertEquals(AbstractMessage.UNSUBSCRIBE, message.getMessageType()); } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java index 343521d5..62301ac8 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java @@ -50,7 +50,7 @@ public class UnsubscribeEncoderTest { //variable part String topic1 = "/topic"; - m_msg.addTopic(topic1); + m_msg.addTopicFilter(topic1); //Exercise m_encoder.encode(m_mockedContext, m_msg, m_out); @@ -76,8 +76,8 @@ public class UnsubscribeEncoderTest { //variable part String topic1 = "a/b"; String topic2 = "a/b/c"; - m_msg.addTopic(topic1); - m_msg.addTopic(topic2); + m_msg.addTopicFilter(topic1); + m_msg.addTopicFilter(topic2); //Exercise m_encoder.encode(m_mockedContext, m_msg, m_out); diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java index c0ccfdf0..29f8aa0a 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java +++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java @@ -27,19 +27,19 @@ public class SubscribeMessage extends MessageIDMessage { public static class Couple { private byte m_qos; - private String m_topic; + private String m_topicFilter; public Couple(byte qos, String topic) { m_qos = qos; - m_topic = topic; + m_topicFilter = topic; } public byte getQos() { return m_qos; } - public String getTopic() { - return m_topic; + public String getTopicFilter() { + return m_topicFilter; } } private List m_subscriptions = new ArrayList(); diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java index d243d50e..c608ac01 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java +++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java @@ -29,11 +29,11 @@ public class UnsubscribeMessage extends MessageIDMessage { m_messageType = AbstractMessage.UNSUBSCRIBE; } - public List topics() { + public List topicFilters() { return m_types; } - public void addTopic(String type) { + public void addTopicFilter(String type) { m_types.add(type); } }