anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Renamed subscribe and unsubscribe topic to topicFilter

Browse Source
This commit is contained in:
selva.andre
2014-08-03 11:17:25 +02:00
parent e40e19cd1e
commit 161421a00c
11 changed files with 23 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+2 -2
View File
@@ -532,8 +532,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
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
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+1 -1
View File
@@ -169,7 +169,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} 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);
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java
+1 -1
View File
@@ -41,7 +41,7 @@ class SubscribeEncoder extends DemuxEncoder<SubscribeMessage> {
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());
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java
+2 -2
View File
@@ -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);
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java
+2 -2
View File
@@ -29,7 +29,7 @@ class UnsubscribeEncoder extends DemuxEncoder<UnsubscribeMessage> {
@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<UnsubscribeMessage> {
ByteBuf buff = null;
try {
variableHeaderBuff.writeShort(message.getMessageID());
for (String topic : message.topics()) {
for (String topic : message.topicFilters()) {
variableHeaderBuff.writeBytes(Utils.encodeString(topic));
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java
+2 -2
View File
@@ -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)
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java
+2 -2
View File
@@ -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());
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java
+2 -3
View File
@@ -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());
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java
+3 -3
View File
@@ -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);
parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
+4 -4
View File
@@ -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<Couple> m_subscriptions = new ArrayList<Couple>();
parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java
+2 -2
View File
@@ -29,11 +29,11 @@ public class UnsubscribeMessage extends MessageIDMessage {
m_messageType = AbstractMessage.UNSUBSCRIBE;
}
public List<String> topics() {
public List<String> topicFilters() {
return m_types;
}
public void addTopic(String type) {
public void addTopicFilter(String type) {
m_types.add(type);
}
}