From 949e5314487532fdb3bf28cf7779926e4dd63d84 Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 10 Jun 2017 16:55:00 +0200 Subject: [PATCH] Switched to Netty 4.1.12 to use MqttMessageBuilders --- .../moquette/server/netty/MessageBuilder.java | 242 ------------------ .../impl/ProtocolProcessorBootstrapper.java | 4 +- .../ServerIntegrationEmbeddedPublishTest.java | 7 +- .../spi/impl/BrokerInterceptorTest.java | 14 +- .../spi/impl/ProtocolProcessorTest.java | 52 ++-- .../impl/ProtocolProcessor_CONNECT_Test.java | 37 ++- .../java/io/moquette/testclient/Client.java | 3 +- .../testembedded/EmbeddedLauncher.java | 22 +- pom.xml | 2 +- 9 files changed, 73 insertions(+), 310 deletions(-) delete mode 100644 broker/src/main/java/io/moquette/server/netty/MessageBuilder.java diff --git a/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java b/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java deleted file mode 100644 index a1746556..00000000 --- a/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java +++ /dev/null @@ -1,242 +0,0 @@ -/* - * Copyright (c) 2012-2017 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ - -package io.moquette.server.netty; - -import io.netty.buffer.Unpooled; -import io.netty.handler.codec.mqtt.*; -import java.util.ArrayList; -import java.util.List; - -public final class MessageBuilder { - - public static class PublishBuilder { - - private String topic; - private boolean retained; - private MqttQoS qos; - private byte[] payload; - private int messageId; - - public PublishBuilder topicName(String topic) { - this.topic = topic; - return PublishBuilder.this; - } - - public PublishBuilder retained(boolean retained) { - this.retained = retained; - return this; - } - - public PublishBuilder qos(MqttQoS qos) { - this.qos = qos; - return this; - } - - public PublishBuilder payload(byte[] payload) { - this.payload = payload; - return this; - } - - public PublishBuilder messageId(int messageId) { - this.messageId = messageId; - return this; - } - - public MqttPublishMessage build() { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0); - MqttPublishVariableHeader mqttVariableHeader = new MqttPublishVariableHeader(topic, messageId); - return new MqttPublishMessage(mqttFixedHeader, mqttVariableHeader, Unpooled.buffer().writeBytes(payload)); - } - } - - public static class ConnectBuilder { - - private MqttVersion version = MqttVersion.MQTT_3_1_1; - private String clientId; - private boolean cleanSession; - private boolean hasUser; - private boolean hasPassword; - private int keepAliveSecs; - private boolean willFlag; - private MqttQoS willQos = MqttQoS.AT_MOST_ONCE; - private String willTopic; - private String willMessage; - private String username; - private String password; - - public ConnectBuilder protocolVersion(MqttVersion version) { - this.version = version; - return this; - } - - public ConnectBuilder clientId(String clientId) { - this.clientId = clientId; - return this; - } - - public ConnectBuilder cleanSession(boolean cleanSession) { - this.cleanSession = cleanSession; - return this; - } - - public ConnectBuilder keepAlive(int keepAliveSecs) { - this.keepAliveSecs = keepAliveSecs; - return this; - } - - public ConnectBuilder willFlag(boolean willFlag) { - this.willFlag = willFlag; - return this; - } - - public ConnectBuilder willQoS(MqttQoS willQos) { - this.willQos = willQos; - return this; - } - - public ConnectBuilder willTopic(String willTopic) { - this.willTopic = willTopic; - return this; - } - - public ConnectBuilder willMessage(String willMessage) { - this.willMessage = willMessage; - return this; - } - - public ConnectBuilder hasUser() { - this.hasUser = true; - return this; - } - - public ConnectBuilder hasPassword() { - this.hasPassword = true; - return this; - } - - public ConnectBuilder username(String username) { - this.username = username; - return this; - } - - public ConnectBuilder password(String password) { - this.password = password; - return this; - } - - public MqttConnectMessage build() { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader( - MqttMessageType.CONNECT, - false, - MqttQoS.AT_MOST_ONCE, - false, - 0); - MqttConnectVariableHeader mqttConnectVariableHeader = new MqttConnectVariableHeader( - version.protocolName(), - version.protocolLevel(), - hasUser, - hasPassword, - false, - willQos.value(), - willFlag, - cleanSession, - keepAliveSecs); - MqttConnectPayload mqttConnectPayload = new MqttConnectPayload( - clientId, - willTopic, - willMessage, - username, - password); - return new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, mqttConnectPayload); - } - } - - public static class SubscribeBuilder { - - private List subscriptions = new ArrayList<>(); - private int messageId; - - public SubscribeBuilder addSubscription(MqttQoS qos, String topic) { - subscriptions.add(new MqttTopicSubscription(topic, qos)); - return this; - } - - public SubscribeBuilder messageId(int messageId) { - this.messageId = messageId; - return this; - } - - public MqttSubscribeMessage build() { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader( - MqttMessageType.SUBSCRIBE, - false, - MqttQoS.AT_LEAST_ONCE, - false, - 0); - MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId); - MqttSubscribePayload mqttSubscribePayload = new MqttSubscribePayload(subscriptions); - return new MqttSubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload); - } - } - - public static class UnsubscribeBuilder { - - private List topicFilters = new ArrayList<>(); - private int messageId; - - public UnsubscribeBuilder addTopicFilter(String topic) { - topicFilters.add(topic); - return this; - } - - public UnsubscribeBuilder messageId(int messageId) { - this.messageId = messageId; - return this; - } - - public MqttUnsubscribeMessage build() { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader( - MqttMessageType.UNSUBSCRIBE, - false, - MqttQoS.AT_LEAST_ONCE, - false, - 0); - MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId); - MqttUnsubscribePayload mqttSubscribePayload = new MqttUnsubscribePayload(topicFilters); - return new MqttUnsubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload); - } - } - - public static ConnectBuilder connect() { - return new ConnectBuilder(); - } - - public static PublishBuilder publish() { - return new PublishBuilder(); - } - - public static SubscribeBuilder subscribe() { - return new SubscribeBuilder(); - } - - public static UnsubscribeBuilder unsubscribe() { - return new UnsubscribeBuilder(); - } - - private MessageBuilder() { - } -} diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 0430b07d..2cc7c892 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -263,8 +263,8 @@ public class ProtocolProcessorBootstrapper { public void shutdown() { if (storeShutdown != null) storeShutdown.run(); - if (m_interceptor != null) - m_interceptor.stop(); +// if (m_interceptor != null) +// m_interceptor.stop(); } public ConnectionDescriptorStore getConnectionDescriptors() { diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index 209b7b7a..d13fc0ba 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -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"); } diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index ffade441..01bb15d4 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -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.asList(new MockObserver())); + Collections.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(); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index ca96cae7..9053bdac 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -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 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); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index c2e71452..011d3b15 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -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); diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java index a98b4d39..8930ded2 100644 --- a/broker/src/test/java/io/moquette/testclient/Client.java +++ b/broker/src/test/java/io/moquette/testclient/Client.java @@ -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(); diff --git a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java index 719da782..0b0bf1db 100644 --- a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java +++ b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java @@ -1,4 +1,4 @@ - /* +/* * Copyright (c) 2012-2017 The original author or authors * ------------------------------------------------------ * All rights reserved. This program and the accompanying materials @@ -23,14 +23,15 @@ import io.moquette.server.config.ClasspathResourceLoader; import io.moquette.server.config.IConfig; import io.moquette.server.config.IResourceLoader; import io.moquette.server.config.ResourceLoaderConfig; -import io.moquette.server.netty.MessageBuilder; +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 java.io.IOException; +import java.util.Collections; import java.util.List; -import static java.util.Arrays.asList; final class EmbeddedLauncher { @@ -53,7 +54,7 @@ final class EmbeddedLauncher { final IConfig classPathConfig = new ResourceLoaderConfig(classpathLoader); final Server mqttBroker = new Server(); - List userHandlers = asList(new PublisherListener()); + List userHandlers = Collections.singletonList(new PublisherListener()); mqttBroker.startServer(classPathConfig, userHandlers); System.out.println("Broker started press [CTRL+C] to stop"); @@ -69,14 +70,15 @@ final class EmbeddedLauncher { Thread.sleep(20000); System.out.println("Before self publish"); - MqttPublishMessage message = MessageBuilder.publish() - .topicName("/exit") - .retained(true) + MqttPublishMessage message = MqttMessageBuilders.publish() + .topicName("/exit") + .retained(true) // qos(MqttQoS.AT_MOST_ONCE); // qQos(MqttQoS.AT_LEAST_ONCE); - .qos(MqttQoS.EXACTLY_ONCE) - .payload("Hello World!!".getBytes()) - .build(); + .qos(MqttQoS.EXACTLY_ONCE) + .payload(Unpooled.copiedBuffer("Hello World!!".getBytes())) + .build(); + mqttBroker.internalPublish(message, "INTRLPUB"); System.out.println("After self publish"); } diff --git a/pom.xml b/pom.xml index 548df226..906a77bd 100644 --- a/pom.xml +++ b/pom.xml @@ -3,7 +3,7 @@ 4.0.0 UTF-8 - 4.1.6.Final + 4.1.12.Final 1.7 1.7 andsel/maven