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

Merge branch 'master' of github.com:andsel/moquette

Browse Source
This commit is contained in:
andsel
2015-09-24 11:31:20 +02:00
parent 5e397ea6d9 4e32988661
commit ea2daa688f
12 changed files with 95 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+3 -3
View File
@@ -224,7 +224,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//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<ValueEvent> {
}
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<ValueEvent> {
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);
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+1 -1
View File
@@ -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);
}
}
broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java
+2 -2
View File
@@ -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);
broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java
+4 -4
View File
@@ -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
netty_parser/src/main/java/org/eclipse/moquette/parser/netty/DemuxDecoder.java
+5 -1
View File
@@ -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;
netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckDecoder.java
+1 -1
View File
@@ -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);
netty_parser/src/main/java/org/eclipse/moquette/parser/netty/SubAckEncoder.java
+1 -2
View File
@@ -40,8 +40,7 @@ class SubAckEncoder extends DemuxEncoder<SubAckMessage> {
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);
netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java
+1 -1
View File
@@ -184,7 +184,7 @@ public class Utils {
flags |= 0x01;
}
flags |= ((message.getQos().ordinal() & 0x03) << 1);
flags |= ((message.getQos().byteValue() & 0x03) << 1);
return flags;
}
netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckDecoderTest.java
+19 -1
View File
@@ -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<AbstractMessage.QOSType> 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());
}
}
}
netty_parser/src/test/java/org/eclipse/moquette/parser/netty/SubAckEncoderTest.java
+25 -4
View File
@@ -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
}
}
parser_commons/src/main/java/org/eclipse/moquette/proto/Utils.java
+1 -1
View File
@@ -173,7 +173,7 @@ public class Utils {
flags |= 0x01;
}
flags |= ((message.getQos().ordinal() & 0x03) << 1);
flags |= ((message.getQos().byteValue() & 0x03) << 1);
return flags;
}
parser_commons/src/main/java/org/eclipse/moquette/proto/messages/AbstractMessage.java
+32 -2
View File
@@ -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