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

Switched to Netty 4.1.12 to use MqttMessageBuilders

Browse Source
This commit is contained in:
andsel
2017-06-10 16:55:00 +02:00
parent f4afd451ab
commit 949e531448
9 changed files with 73 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java
+4 -3
View File
@@ -16,9 +16,10 @@
package io.moquette.server;
import io.moquette.server.netty.MessageBuilder;
import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.eclipse.paho.client.mqttv3.*;
@@ -84,8 +85,8 @@ public class ServerIntegrationEmbeddedPublishTest {
}
private void internalPublishToWithQosAndRetained(String topic, MqttQoS qos, boolean retained) {
MqttPublishMessage message = MessageBuilder.publish().topicName(topic).retained(retained).qos(qos)
.payload("Hello world MQTT!!".getBytes()).build();
MqttPublishMessage message = MqttMessageBuilders.publish().topicName(topic).retained(retained).qos(qos)
.payload(Unpooled.copiedBuffer("Hello world MQTT!!".getBytes())).build();
m_server.internalPublish(message, "INTRPUBL");
}
broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java
+9 -5
View File
@@ -18,15 +18,18 @@ package io.moquette.spi.impl;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.*;
import io.moquette.server.netty.MessageBuilder;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.refEq;
import static org.mockito.Mockito.mock;
@@ -87,7 +90,7 @@ public class BrokerInterceptorTest {
}
private static final BrokerInterceptor interceptor = new BrokerInterceptor(
Arrays.<InterceptHandler>asList(new MockObserver()));
Collections.<InterceptHandler>singletonList(new MockObserver()));
@BeforeClass
public static void beforeAllTests() {
@@ -107,7 +110,7 @@ public class BrokerInterceptorTest {
@Test
public void testNotifyClientConnected() throws Exception {
interceptor.notifyClientConnected(MessageBuilder.connect().build());
interceptor.notifyClientConnected(MqttMessageBuilders.connect().build());
interval();
assertEquals(40, n.get());
}
@@ -122,7 +125,8 @@ public class BrokerInterceptorTest {
@Test
public void testNotifyTopicPublished() throws Exception {
interceptor.notifyTopicPublished(
MessageBuilder.publish().qos(MqttQoS.AT_MOST_ONCE).payload("Hello".getBytes()).build(),
MqttMessageBuilders.publish().qos(MqttQoS.AT_MOST_ONCE)
.payload(Unpooled.copiedBuffer("Hello".getBytes())).build(),
"cli1234",
"cli1234");
interval();
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+26 -26
View File
@@ -18,7 +18,6 @@ package io.moquette.spi.impl;
import io.moquette.interception.InterceptHandler;
import io.moquette.persistence.MemoryStorageService;
import io.moquette.server.netty.MessageBuilder;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
@@ -29,6 +28,7 @@ import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.*;
import org.junit.Before;
@@ -70,7 +70,7 @@ public class ProtocolProcessorTest {
/*
* connMsg = new ConnectMessage(); connMsg.setProtocolVersion((byte) 0x03);
*
* connMsg = MessageBuilder.connect() .protocolVersion(MqttVersion.MQTT_3_1)
* connMsg = MqttMessageBuilders.connect() .protocolVersion(MqttVersion.MQTT_3_1)
* .clientId(FAKE_CLIENT_ID) .cleanSession(true) .build();
*/
@@ -127,13 +127,13 @@ public class ProtocolProcessorTest {
true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage connectMessage = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
MqttConnectMessage connectMessage = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
m_processor.processConnect(m_channel, connectMessage);
// Exercise
MqttPublishMessage msg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.retained(false).payload("Hello".getBytes()).build();
MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes())).build();
NettyUtils.userName(m_channel, "FakeCLI");
m_processor.processPublish(m_channel, msg);
@@ -171,21 +171,21 @@ public class ProtocolProcessorTest {
NO_OBSERVERS_INTERCEPTOR);
EmbeddedChannel firstReceiverChannel = new EmbeddedChannel();
MqttConnectMessage connectMessage = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
MqttConnectMessage connectMessage = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
m_processor.processConnect(firstReceiverChannel, connectMessage);
assertConnAckAccepted(firstReceiverChannel);
// connect the second fake subscriber
EmbeddedChannel secondReceiverChannel = new EmbeddedChannel();
MqttConnectMessage connectMessage2 = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
MqttConnectMessage connectMessage2 = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID2).cleanSession(true).build();
m_processor.processConnect(secondReceiverChannel, connectMessage2);
assertConnAckAccepted(secondReceiverChannel);
// Exercise
MqttPublishMessage msg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.retained(false).payload("Hello".getBytes()).build();
MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes())).build();
NettyUtils.userName(m_channel, "FakeCLI");
m_processor.processPublish(m_channel, msg);
@@ -205,7 +205,7 @@ public class ProtocolProcessorTest {
@Test
public void testSubscribe() {
MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
.messageId(10).build();
// Exercise
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
@@ -231,7 +231,7 @@ public class ProtocolProcessorTest {
NO_OBSERVERS_INTERCEPTOR);
// Exercise
MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
.messageId(10).build();
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
@@ -252,7 +252,7 @@ public class ProtocolProcessorTest {
@Test
public void testDoubleSubscribe() {
MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
.messageId(10).build();
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
assertEquals(0, subscriptions.size());
@@ -273,7 +273,7 @@ public class ProtocolProcessorTest {
@Test
public void testSubscribeWithBadFormattedTopic() {
MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, BAD_FORMATTED_TOPIC)
MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, BAD_FORMATTED_TOPIC)
.messageId(10).build();
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
@@ -294,7 +294,7 @@ public class ProtocolProcessorTest {
*/
@Test
public void testUnsubscribeWithBadFormattedTopic() {
MqttUnsubscribeMessage msg = MessageBuilder.unsubscribe().addTopicFilter(BAD_FORMATTED_TOPIC).messageId(1)
MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe().addTopicFilter(BAD_FORMATTED_TOPIC).messageId(1)
.build();
// Exercise
@@ -328,20 +328,20 @@ public class ProtocolProcessorTest {
// simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage connectMessage = MessageBuilder.connect().clientId(FAKE_PUBLISHER_ID)
MqttConnectMessage connectMessage = MqttMessageBuilders.connect().clientId(FAKE_PUBLISHER_ID)
.protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true).build();
m_processor.processConnect(m_channel, connectMessage);
assertConnAckAccepted(m_channel);
MqttPublishMessage pubmsg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.payload("Hello".getBytes()).retained(true).build();
MqttPublishMessage pubmsg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).build();
NettyUtils.clientID(m_channel, FAKE_PUBLISHER_ID);
m_processor.processPublish(m_channel, pubmsg);
NettyUtils.cleanSession(m_channel, false);
// Exercise
MqttSubscribeMessage msg = MessageBuilder.subscribe().messageId(10).addSubscription(MqttQoS.AT_MOST_ONCE, "#")
MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().messageId(10).addSubscription(MqttQoS.AT_MOST_ONCE, "#")
.build();
m_processor.processSubscribe(m_channel, msg);
@@ -367,7 +367,7 @@ public class ProtocolProcessorTest {
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage connectMessage = MessageBuilder.connect().clientId(FAKE_PUBLISHER_ID)
MqttConnectMessage connectMessage = MqttMessageBuilders.connect().clientId(FAKE_PUBLISHER_ID)
.protocolVersion(MqttVersion.MQTT_3_1).build();
m_processor.processConnect(m_channel, connectMessage);
@@ -391,8 +391,8 @@ public class ProtocolProcessorTest {
NO_OBSERVERS_INTERCEPTOR);
// Exercise
MqttPublishMessage msg = MessageBuilder.publish().topicName("/topic").qos(MqttQoS.AT_MOST_ONCE)
.payload("Hello".getBytes()).retained(true).build();
MqttPublishMessage msg = MqttMessageBuilders.publish().topicName("/topic").qos(MqttQoS.AT_MOST_ONCE)
.payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).build();
NettyUtils.clientID(m_channel, "Publisher");
m_processor.processPublish(m_channel, msg);
@@ -408,13 +408,13 @@ public class ProtocolProcessorTest {
@Test
public void testCleanRetainedStoreAfterAQoS0AndRetainedTrue() {
// force a connect
connMsg = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1).clientId("Publisher")
connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).clientId("Publisher")
.cleanSession(true).build();
m_processor.processConnect(m_channel, connMsg);
// prepare and existing retained store
NettyUtils.clientID(m_channel, "Publisher");
MqttPublishMessage msg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_LEAST_ONCE)
.payload("Hello".getBytes()).retained(true).messageId(100).build();
MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_LEAST_ONCE)
.payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).messageId(100).build();
m_processor.processPublish(m_channel, msg);
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
@@ -426,8 +426,8 @@ public class ProtocolProcessorTest {
assertFalse(messages.isEmpty());
// Exercise
MqttPublishMessage cleanPubMsg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.payload("Hello".getBytes()).retained(true).build();
MqttPublishMessage cleanPubMsg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).build();
m_processor.processPublish(m_channel, cleanPubMsg);
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+18 -19
View File
@@ -17,7 +17,6 @@
package io.moquette.spi.impl;
import io.moquette.persistence.MemoryStorageService;
import io.moquette.server.netty.MessageBuilder;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
@@ -27,16 +26,15 @@ import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.handler.codec.mqtt.*;
import org.junit.Before;
import org.junit.Test;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
@@ -46,7 +44,7 @@ import static org.junit.Assert.assertTrue;
public class ProtocolProcessor_CONNECT_Test {
EmbeddedChannel m_session;
MessageBuilder.ConnectBuilder connMsg;
MqttMessageBuilders.ConnectBuilder connMsg;
ProtocolProcessor m_processor;
IMessagesStore m_messagesStore;
@@ -56,7 +54,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Before
public void setUp() throws InterruptedException {
connMsg = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
m_session = new EmbeddedChannel();
@@ -130,7 +128,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void validAuthentication() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).hasUser().hasPassword()
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER).password(ProtocolProcessorTest.TEST_PWD).build();
// Exercise
@@ -143,7 +141,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void noPasswdAuthentication() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).hasUser()
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER).build();
// Exercise
@@ -156,7 +154,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void invalidAuthentication() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).hasUser().hasPassword()
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER + "_fake").password(ProtocolProcessorTest.TEST_PWD).build();
// Exercise
@@ -189,7 +187,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void prohibitAnonymousClient_providingUsername() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).hasUser()
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER + "_fake").build();
m_processor.init(
subscriptions,
@@ -248,7 +246,7 @@ public class ProtocolProcessor_CONNECT_Test {
// second connect with clean session false
m_session = new EmbeddedChannel();
MqttConnectMessage secondConnMsg = MessageBuilder.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1).build();
m_processor.processConnect(m_session, secondConnMsg);
@@ -265,15 +263,16 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() {
// Connect a client1
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).hasUser().hasPassword()
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER).password(ProtocolProcessorTest.TEST_PWD).build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
// create another connect same clientID but with bad credentials
MqttConnectMessage evilClientConnMsg = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).hasUser().username(ProtocolProcessorTest.EVIL_TEST_USER)
.hasPassword().password(ProtocolProcessorTest.EVIL_TEST_PWD).build();
MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.EVIL_TEST_USER)
.password(ProtocolProcessorTest.EVIL_TEST_PWD).build();
EmbeddedChannel evilSession = new EmbeddedChannel();
@@ -312,14 +311,14 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void testMultipleReconnection() throws InterruptedException {
// connect with clean a false and subscribe to a topic
MqttConnectMessage msg = MessageBuilder.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1_1).build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
// subscribe
MqttSubscribeMessage subscribeMsg = MessageBuilder.subscribe()
MqttSubscribeMessage subscribeMsg = MqttMessageBuilders.subscribe()
.addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.FAKE_TOPIC).messageId(10).build();
NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID);
@@ -361,7 +360,7 @@ public class ProtocolProcessor_CONNECT_Test {
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
// Connect message without clean session set to true but client id is still null
MqttConnectMessage msg = MessageBuilder.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1)
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1)
.build();
m_processor.processConnect(m_session, msg);
broker/src/test/java/io/moquette/testclient/Client.java
+1 -2
View File
@@ -17,7 +17,6 @@
package io.moquette.testclient;
import io.moquette.BrokerConstants;
import io.moquette.server.netty.MessageBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
@@ -126,7 +125,7 @@ public class Client {
}
public void connect() {
MqttConnectMessage connectMessage = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1_1)
MqttConnectMessage connectMessage = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1_1)
.clientId("").keepAlive(2) // secs
.willFlag(false).willQoS(MqttQoS.AT_MOST_ONCE).build();