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

Minor, modified access to fields of SubscribeMessage.Couple

Browse Source
This commit is contained in:
andsel
2016-01-30 17:35:28 +01:00
parent a19af18a61
commit d36de24349
7 changed files with 25 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+4
View File
@@ -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
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+5 -5
View File
@@ -659,16 +659,16 @@ public class ProtocolProcessor {
String user = NettyUtils.userName(channel);
List<Subscription> 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);
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+2 -10
View File
@@ -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);
netty_parser/src/main/java/io/moquette/parser/netty/SubscribeEncoder.java
+2 -2
View File
@@ -41,8 +41,8 @@ class SubscribeEncoder extends DemuxEncoder<SubscribeMessage> {
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();
netty_parser/src/test/java/io/moquette/parser/netty/SubscribeDecoderTest.java
+4 -4
View File
@@ -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()));
netty_parser/src/test/java/io/moquette/parser/netty/SubscribeEncoderTest.java
+4 -4
View File
@@ -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)
parser_commons/src/main/java/io/moquette/proto/messages/SubscribeMessage.java
+4 -11
View File
@@ -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<Couple> m_subscriptions = new ArrayList<>();