From ba8ad0228930d04479c714f5ab9b5c1ff10bdb92 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jul 2018 17:57:52 +0200 Subject: [PATCH] Dropper Hazelcast because Moquette is not designed to be clusterizable Separated the connect/disconnect logic into ConnectHandler/DisconnectHandler Moved the sending of CONN ACK after session creation/update Moved the will message management inside the session class --- ChangeLog.txt | 4 + broker/build.gradle | 1 - .../java/io/moquette/BrokerConstants.java | 1 - .../connections/IConnectionsManager.java | 4 +- .../HazelcastInterceptHandler.java | 56 --- .../moquette/interception/HazelcastMsg.java | 53 --- .../moquette/server/ConnectionDescriptor.java | 16 +- .../server/ConnectionDescriptorStore.java | 5 - .../io/moquette/server/HazelcastListener.java | 62 --- .../main/java/io/moquette/server/Server.java | 54 --- .../java/io/moquette/spi/ClientSession.java | 12 +- .../java/io/moquette/spi/DurableSession.java | 37 +- .../java/io/moquette/spi/ISessionsStore.java | 1 + .../io/moquette/spi/TransientSession.java | 38 +- .../io/moquette/spi/impl/ConnectHandler.java | 285 ++++++++++++ .../moquette/spi/impl/DisconnectHandler.java | 133 ++++++ .../spi/impl/InternalRepublisher.java | 1 + .../moquette/spi/impl/ProtocolProcessor.java | 409 ++---------------- .../moquette/spi/impl/SessionsRepository.java | 8 +- .../io/moquette/spi/impl/WillMessage.java | 36 ++ ...rationHazelcastHandlerInterceptorTest.java | 153 ------- .../moquette/spi/impl/ConnectHandlerTest.java | 315 ++++++++++++++ .../impl/ProtocolProcessor_CONNECT_Test.java | 326 ++------------ ...rotocolProcessor_VariousUseCases_Test.java | 29 +- 24 files changed, 944 insertions(+), 1095 deletions(-) delete mode 100644 broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java delete mode 100644 broker/src/main/java/io/moquette/interception/HazelcastMsg.java delete mode 100644 broker/src/main/java/io/moquette/server/HazelcastListener.java create mode 100644 broker/src/main/java/io/moquette/spi/impl/ConnectHandler.java create mode 100644 broker/src/main/java/io/moquette/spi/impl/DisconnectHandler.java create mode 100644 broker/src/main/java/io/moquette/spi/impl/WillMessage.java delete mode 100644 broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java create mode 100644 broker/src/test/java/io/moquette/spi/impl/ConnectHandlerTest.java diff --git a/ChangeLog.txt b/ChangeLog.txt index a4b5f9b5..3a84a44f 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,3 +1,7 @@ +Version 0.12: + Fixed issue #415 + Dropped Hazelcast because at the moment Moquette is not designed to be clusterizable + Version 0.10: Netty epoll system added Separated in memory and MapDB storage implementations diff --git a/broker/build.gradle b/broker/build.gradle index 439ef515..af90eb96 100644 --- a/broker/build.gradle +++ b/broker/build.gradle @@ -24,7 +24,6 @@ dependencies { compile group: 'commons-codec', name: 'commons-codec', version:'1.10' // compile group: 'org.hdrhistogram', name: 'HdrHistogram', version:'2.1.4' - compile group: 'com.hazelcast', name: 'hazelcast', version:'3.5.4' compile group: 'org.slf4j', name: 'slf4j-api', version:'1.7.5' testCompile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 5eee7edc..ffa93727 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -60,7 +60,6 @@ public final class BrokerConstants { public static final String DISABLED_PORT_BIND = "disabled"; public static final String HOST = "0.0.0.0"; public static final String NEED_CLIENT_AUTH = "need_client_auth"; - public static final String HAZELCAST_CONFIGURATION = "hazelcast.configuration"; public static final String NETTY_SO_BACKLOG_PROPERTY_NAME = "netty.so_backlog"; public static final String NETTY_SO_REUSEADDR_PROPERTY_NAME = "netty.so_reuseaddr"; public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay"; diff --git a/broker/src/main/java/io/moquette/connections/IConnectionsManager.java b/broker/src/main/java/io/moquette/connections/IConnectionsManager.java index 90e6be28..0a1cafea 100644 --- a/broker/src/main/java/io/moquette/connections/IConnectionsManager.java +++ b/broker/src/main/java/io/moquette/connections/IConnectionsManager.java @@ -39,10 +39,8 @@ public interface IConnectionsManager { boolean removeConnection(ConnectionDescriptor descriptor); - ConnectionDescriptor getConnection(String clientID); - /** - * Determines wether a MQTT client is connected to the broker. + * Determines weather a MQTT client is connected to the broker. * * @param clientID * @return diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java deleted file mode 100644 index 3d2908ee..00000000 --- a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.interception; - -import com.hazelcast.core.HazelcastInstance; -import com.hazelcast.core.ITopic; -import io.moquette.interception.messages.InterceptPublishMessage; -import io.moquette.server.Server; -import io.netty.buffer.ByteBuf; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import static io.moquette.spi.impl.Utils.readBytesAndRewind; -import static java.nio.charset.StandardCharsets.UTF_8; - -public class HazelcastInterceptHandler extends AbstractInterceptHandler { - - private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class); - private final HazelcastInstance hz; - - public HazelcastInterceptHandler(Server server) { - this.hz = server.getHazelcastInstance(); - } - - @Override - public String getID() { - return HazelcastInterceptHandler.class.getName() + "@" + hz.getName(); - } - - @Override - public void onPublish(InterceptPublishMessage msg) { - // TODO ugly, too much array copy - ByteBuf payload = msg.getPayload(); - byte[] payloadContent = readBytesAndRewind(payload); - - LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), - new String(payloadContent, UTF_8)); - ITopic topic = hz.getTopic("moquette"); - HazelcastMsg hazelcastMsg = new HazelcastMsg(msg); - topic.publish(hazelcastMsg); - } - -} diff --git a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java deleted file mode 100644 index 3af14642..00000000 --- a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.interception; - -import io.moquette.interception.messages.InterceptPublishMessage; -import java.io.Serializable; -import static io.moquette.spi.impl.Utils.readBytesAndRewind; - -public class HazelcastMsg implements Serializable { - - private static final long serialVersionUID = -1431584750134928273L; - private final String clientId; - private final int qos; - private final byte[] payload; - private final String topic; - - public HazelcastMsg(InterceptPublishMessage msg) { - this.clientId = msg.getClientID(); - this.topic = msg.getTopicName(); - this.qos = msg.getQos().value(); - this.payload = readBytesAndRewind(msg.getPayload()); - } - - public String getClientId() { - return clientId; - } - - public int getQos() { - return qos; - } - - public byte[] getPayload() { - return payload; - } - - public String getTopic() { - return topic; - } -} diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 46d6f7c9..75831122 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -22,6 +22,7 @@ import io.moquette.server.netty.metrics.BytesMetricsHandler; import io.moquette.server.netty.metrics.MessageMetrics; import io.moquette.server.netty.metrics.MessageMetricsHandler; import io.netty.channel.Channel; +import io.netty.channel.ChannelFutureListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,8 +43,8 @@ public class ConnectionDescriptor { public enum ConnectionState { // Connection states DISCONNECTED, + INIT_SESSION, SENDACK, - SESSION_CREATED, MESSAGES_REPUBLISHED, ESTABLISHED, // Disconnection states @@ -67,6 +68,10 @@ public class ConnectionDescriptor { this.channel.writeAndFlush(payload).addListener(FIRE_EXCEPTION_ON_FAILURE); } + public void writeAndFlush(Object payload, ChannelFutureListener listener) { + this.channel.writeAndFlush(payload).addListener(listener); + } + public boolean doesNotUseChannel(Channel channel) { return !(this.channel.equals(channel)); } @@ -86,13 +91,8 @@ public class ConnectionDescriptor { } public void abort() { - LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID); - // try { - // this.channel.disconnect().sync(); - this.channel.close().addListener(CLOSE_ON_FAILURE); // .sync(); - // } catch (InterruptedException e) { - // e.printStackTrace(); - // } + LOG.info("Closing connection descriptor clientId = {}.", clientID); + this.channel.close().addListener(CLOSE_ON_FAILURE); } public boolean assignState(ConnectionState expected, ConnectionState newState) { diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java index 0b207345..e043e78e 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java @@ -62,11 +62,6 @@ public class ConnectionDescriptorStore implements IConnectionsManager { return connectionDescriptors.remove(descriptor.clientID, descriptor); } - @Override - public ConnectionDescriptor getConnection(String clientID) { - return connectionDescriptors.get(clientID); - } - @Override public boolean isConnected(String clientID) { return connectionDescriptors.containsKey(clientID); diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java deleted file mode 100644 index 5e8cce7f..00000000 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright (c) 2012-2018 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; - -import com.hazelcast.core.Message; -import com.hazelcast.core.MessageListener; -import io.moquette.interception.HazelcastMsg; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.handler.codec.mqtt.*; -import io.netty.util.ReferenceCountUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class HazelcastListener implements MessageListener { - - private static final Logger LOG = LoggerFactory.getLogger(HazelcastListener.class); - - private final Server server; - - public HazelcastListener(Server server) { - this.server = server; - } - - @Override - public void onMessage(Message msg) { - ByteBuf payload = null; - try { - if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { - HazelcastMsg hzMsg = msg.getMessageObject(); - LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(), hzMsg.getTopic(), - hzMsg.getPayload()); - // TODO pass forward this information in somehow publishMessage.setLocal(false); - - MqttQoS qos = MqttQoS.valueOf(hzMsg.getQos()); - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0); - MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(hzMsg.getTopic(), 0); - payload = Unpooled.wrappedBuffer(hzMsg.getPayload()); - MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, payload); - server.internalPublish(publishMessage, hzMsg.getClientId()); - } - } catch (Exception ex) { - LOG.error("error polling hazelcast msg queue", ex); - } finally { - ReferenceCountUtil.release(payload); - } - } -} diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 55b794cb..e053dc2f 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -16,17 +16,8 @@ package io.moquette.server; -import com.hazelcast.config.ClasspathXmlConfig; -import com.hazelcast.config.Config; -import com.hazelcast.config.FileSystemXmlConfig; -import com.hazelcast.core.Hazelcast; -import com.hazelcast.core.HazelcastInstance; -import com.hazelcast.core.HazelcastInstanceNotActiveException; -import com.hazelcast.core.ITopic; import io.moquette.BrokerConstants; import io.moquette.connections.IConnectionsManager; -import io.moquette.interception.HazelcastInterceptHandler; -import io.moquette.interception.HazelcastMsg; import io.moquette.interception.InterceptHandler; import io.moquette.server.config.*; import io.moquette.server.netty.NettyAcceptor; @@ -41,7 +32,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; -import java.io.FileNotFoundException; import java.io.IOException; import java.util.Collections; import java.util.List; @@ -58,15 +48,12 @@ public class Server { private static final Logger LOG = LoggerFactory.getLogger(Server.class); - private static final String HZ_INTERCEPT_HANDLER = HazelcastInterceptHandler.class.getCanonicalName(); - private ServerAcceptor m_acceptor; private volatile boolean m_initialized; private ProtocolProcessor m_processor; - private HazelcastInstance hazelcastInstance; private ProtocolProcessorBootstrapper m_processorBootstrapper; @@ -167,7 +154,6 @@ public class Server { if (handlerProp != null) { config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp); } - configureCluster(config); final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME); LOG.debug("Configuring Using persistent store file, path={}", persistencePath); m_processorBootstrapper = new ProtocolProcessorBootstrapper(); @@ -188,38 +174,6 @@ public class Server { m_initialized = true; } - private void configureCluster(IConfig config) throws FileNotFoundException { - LOG.info("Configuring embedded Hazelcast instance"); - String interceptHandlerClassname = config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); - if (interceptHandlerClassname == null || !HZ_INTERCEPT_HANDLER.equals(interceptHandlerClassname)) { - LOG.info("There are no Hazelcast intercept handlers. The server won't start a Hazelcast instance."); - return; - } - String hzConfigPath = config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION); - if (hzConfigPath != null) { - boolean isHzConfigOnClasspath = this.getClass().getClassLoader().getResource(hzConfigPath) != null; - Config hzconfig = isHzConfigOnClasspath - ? new ClasspathXmlConfig(hzConfigPath) - : new FileSystemXmlConfig(hzConfigPath); - LOG.info("Starting Hazelcast instance. ConfigurationFile={}", hzconfig); - hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig); - } else { - LOG.info("Starting Hazelcast instance with default configuration"); - hazelcastInstance = Hazelcast.newHazelcastInstance(); - } - listenOnHazelCastMsg(); - } - - private void listenOnHazelCastMsg() { - LOG.info("Subscribing to Hazelcast topic. TopicName={}", "moquette"); - HazelcastInstance hz = getHazelcastInstance(); - ITopic topic = hz.getTopic("moquette"); - topic.addMessageListener(new HazelcastListener(this)); - } - - public HazelcastInstance getHazelcastInstance() { - return hazelcastInstance; - } /** * Use the broker to publish a message. It's intended for embedding applications. It can be used @@ -246,14 +200,6 @@ public class Server { LOG.trace("Stopping MQTT protocol processor"); m_processorBootstrapper.shutdown(); m_initialized = false; - if (hazelcastInstance != null) { - LOG.trace("Stopping embedded Hazelcast instance"); - try { - hazelcastInstance.shutdown(); - } catch (HazelcastInstanceNotActiveException e) { - LOG.warn("embedded Hazelcast instance is already shut down."); - } - } // calling shutdown() does not actually stop tasks that are not cancelled, // and SessionsRepository does not stop its tasks. Thus shutdownNow(). diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 52ac63d6..11cc8851 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -16,8 +16,10 @@ package io.moquette.spi; +import io.moquette.spi.impl.WillMessage; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; +import io.netty.handler.codec.mqtt.MqttConnectMessage; import java.util.HashSet; import java.util.Set; @@ -111,10 +113,16 @@ public abstract class ClientSession { public abstract int getInflightMessagesNo(); + public abstract void wipeSubscriptions(); + + public abstract void storeWillMessage(MqttConnectMessage msg, String clientId); + + public abstract void removeWill(); + + public abstract WillMessage willMessage(); + @Override public String toString() { return getClass().getSimpleName() + "{clientID='" + clientID + "'}"; } - - public abstract void wipeSubscriptions(); } diff --git a/broker/src/main/java/io/moquette/spi/DurableSession.java b/broker/src/main/java/io/moquette/spi/DurableSession.java index 14fabb5f..467c4877 100644 --- a/broker/src/main/java/io/moquette/spi/DurableSession.java +++ b/broker/src/main/java/io/moquette/spi/DurableSession.java @@ -15,14 +15,20 @@ */ package io.moquette.spi; +import io.moquette.spi.impl.WillMessage; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; +import io.netty.handler.codec.mqtt.MqttConnectMessage; +import io.netty.handler.codec.mqtt.MqttQoS; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.Collection; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.ConcurrentMap; public class DurableSession extends ClientSession { @@ -56,11 +62,14 @@ public class DurableSession extends ClientSession { private final OutboundFlightZone outboundFlightZone; private final InboundFlightZone inboundFlightZone; + private final ConcurrentMap willMessages; - public DurableSession(String clientID, ISessionsStore sessions, ISubscriptionsStore subscriptionsStore) { + public DurableSession(String clientID, ISessionsStore sessions, ISubscriptionsStore subscriptionsStore, + ConcurrentMap willMessages) { super(clientID); this.subscriptionsStore = subscriptionsStore; this.sessionsStore = sessions; + this.willMessages = willMessages; this.outboundFlightZone = new OutboundFlightZone(); this.inboundFlightZone = new InboundFlightZone(); } @@ -215,4 +224,30 @@ public class DurableSession extends ClientSession { public void wipeSubscriptions() { this.subscriptionsStore.wipeSubscriptions(clientID); } + + @Override + public void storeWillMessage(MqttConnectMessage msg, String clientId) { + // Handle will flag + if (msg.variableHeader().isWillFlag()) { + MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos()); + LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}", + clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain()); + byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8); + ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); + // save the will testament in the clientID store + WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos); + willMessages.put(clientId, will); + LOG.debug("MQTT last will and testament has been configured. CId={}", clientId); + } + } + + @Override + public void removeWill() { + willMessages.remove(clientID); + } + + @Override + public WillMessage willMessage() { + return willMessages.get(clientID); + } } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index a50d94ee..3c253f23 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -18,6 +18,7 @@ package io.moquette.spi; import io.moquette.persistence.PersistentSession; import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.impl.WillMessage; import java.time.LocalDateTime; import java.util.Collection; diff --git a/broker/src/main/java/io/moquette/spi/TransientSession.java b/broker/src/main/java/io/moquette/spi/TransientSession.java index f775fff3..b2f1a435 100644 --- a/broker/src/main/java/io/moquette/spi/TransientSession.java +++ b/broker/src/main/java/io/moquette/spi/TransientSession.java @@ -16,11 +16,17 @@ package io.moquette.spi; import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.impl.ProtocolProcessor; +import io.moquette.spi.impl.WillMessage; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; +import io.netty.handler.codec.mqtt.MqttConnectMessage; +import io.netty.handler.codec.mqtt.MqttQoS; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -37,9 +43,12 @@ public class TransientSession extends ClientSession { private final ConcurrentMap inboundInflightMap = new ConcurrentHashMap<>(); private final ConcurrentMap outboundInflightMap = new ConcurrentHashMap<>(); private final Map secondPhaseStore = new ConcurrentHashMap<>(); + private WillMessage will; + private final ConcurrentMap willMessages; - public TransientSession(String clientID) { + public TransientSession(String clientID, ConcurrentMap willMessages) { super(clientID); + this.willMessages = willMessages; } @Override @@ -186,4 +195,31 @@ public class TransientSession extends ClientSession { public void wipeSubscriptions() { // it's transient, no need to wide subs because wasn't stored } + + @Override + public void storeWillMessage(MqttConnectMessage msg, String clientId) { + // Handle will flag + if (msg.variableHeader().isWillFlag()) { + MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos()); + LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}", + clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain()); + byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8); + ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); + // save the will testament in the clientID store + will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos); + willMessages.put(clientId, will); + LOG.debug("MQTT last will and testament has been configured. CId={}", clientId); + } + } + + @Override + public void removeWill() { + willMessages.remove(clientID); + will = null; + } + + @Override + public WillMessage willMessage() { + return will; + } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ConnectHandler.java b/broker/src/main/java/io/moquette/spi/impl/ConnectHandler.java new file mode 100644 index 00000000..17f92c65 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/ConnectHandler.java @@ -0,0 +1,285 @@ +package io.moquette.spi.impl; + +import io.moquette.connections.IConnectionsManager; +import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.netty.AutoFlushHandler; +import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.ClientSession; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; +import io.moquette.spi.security.IAuthenticator; +import io.moquette.spi.security.IAuthorizator; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelPipeline; +import io.netty.handler.codec.mqtt.*; +import io.netty.handler.timeout.IdleStateHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.NoSuchElementException; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; +import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE; +import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; + +public class ConnectHandler { + + private static final Logger LOG = LoggerFactory.getLogger(ConnectHandler.class); + + private IConnectionsManager connectionDescriptors; + private BrokerInterceptor m_interceptor; + private SessionsRepository sessionsRepository; + private ISessionsStore m_sessionsStore; + private IAuthenticator m_authenticator; + private IAuthorizator m_authorizator; + private ISubscriptionsDirectory subscriptions; + private boolean allowAnonymous; + private boolean allowZeroByteClientId; + private boolean reauthorizeSubscriptionsOnConnect; + private InternalRepublisher internalRepublisher; + + ConnectHandler(IConnectionsManager connectedClients, BrokerInterceptor interceptor, SessionsRepository sessionsRepository, + ISessionsStore sessionsStore, IAuthenticator authenticator, IAuthorizator authorizator, + ISubscriptionsDirectory subscriptions, boolean allowAnonymous, boolean allowZeroByteClientId, + boolean reauthorizeSubscriptionsOnConnect, InternalRepublisher internalRepublisher) { + this.connectionDescriptors = connectedClients; + this.m_interceptor = interceptor; + this.sessionsRepository = sessionsRepository; + this.m_sessionsStore = sessionsStore; + this.m_authenticator = authenticator; + this.m_authorizator = authorizator; + this.subscriptions = subscriptions; + this.allowAnonymous = allowAnonymous; + this.allowZeroByteClientId = allowZeroByteClientId; + this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect; + this.internalRepublisher = internalRepublisher; + } + + public void processConnect(Channel channel, MqttConnectMessage msg) { + MqttConnectPayload payload = msg.payload(); + String clientId = payload.clientIdentifier(); + final String username = payload.userName(); + LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, username); + + if (isNotProtocolVersion(msg, MqttVersion.MQTT_3_1) && isNotProtocolVersion(msg, MqttVersion.MQTT_3_1_1)) { + MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION); + + LOG.error("MQTT protocol version is not valid. CId={}", clientId); + channel.writeAndFlush(badProto).addListener(FIRE_EXCEPTION_ON_FAILURE); + channel.close().addListener(CLOSE_ON_FAILURE); + return; + } + + final boolean cleanSession = msg.variableHeader().isCleanSession(); + if (clientId == null || clientId.length() == 0) { + if (!cleanSession || !this.allowZeroByteClientId) { + MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED); + + channel.writeAndFlush(badId).addListener(FIRE_EXCEPTION_ON_FAILURE); + channel.close().addListener(CLOSE_ON_FAILURE); + LOG.error("MQTT client ID cannot be empty. Username={}", username); + return; + } + + // Generating client id. + clientId = UUID.randomUUID().toString().replace("-", ""); + LOG.info("Client has connected with server generated id={}, username={}", clientId, username); + } + + if (!login(channel, msg, clientId)) { + channel.close().addListener(CLOSE_ON_FAILURE); + return; + } + + ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession); + final ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor); + if (existing != null) { + LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId); + existing.abort(); + //return; + this.connectionDescriptors.removeConnection(existing); + this.connectionDescriptors.addConnection(descriptor); + } + + if (!descriptor.assignState(DISCONNECTED, INIT_SESSION)) { + channel.close().addListener(CLOSE_ON_FAILURE); + return; + } + + LOG.debug("Initializing client session {}", clientId); + ClientSession existingSession = this.sessionsRepository.sessionForClient(clientId); + boolean isSessionAlreadyStored = existingSession != null; + final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession(); + if (isSessionAlreadyStored && msgCleanSessionFlag) { + for (Subscription existingSub : existingSession.getSubscriptions()) { + this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientId); + } + } + + initializeKeepAliveTimeout(channel, msg, clientId); + final ClientSession clientSession = this.sessionsRepository.createOrLoadClientSession(clientId, cleanSession); + clientSession.storeWillMessage(msg, clientId); + int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */; + setupAutoFlusher(channel, flushIntervalMs); + + if (!cleanSession && reauthorizeSubscriptionsOnConnect) { + reauthorizeOnExistingSubscriptions(clientId, username); + } + + if (!descriptor.assignState(INIT_SESSION, SENDACK)) { + channel.close().addListener(CLOSE_ON_FAILURE); + return; + } + LOG.debug("Sending CONNACK. CId={}", clientId); + MqttConnAckMessage okResp = createConnectAck(msg, clientId); + + final String connectClientId = clientId; + + descriptor.writeAndFlush(okResp, new ChannelFutureListener() { + @Override + public void operationComplete(ChannelFuture future) { + if (future.isSuccess()) { + LOG.debug("CONNACK has been sent. CId={}", connectClientId); + + if (!descriptor.assignState(SENDACK, MESSAGES_REPUBLISHED)) { + channel.close().addListener(CLOSE_ON_FAILURE); + return; + } + m_interceptor.notifyClientConnected(msg); + if (!msg.variableHeader().isCleanSession()) { + // force the republish of stored QoS1 and QoS2 + internalRepublisher.publishStored(clientSession); + } + + if (!descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED)) { + channel.close().addListener(CLOSE_ON_FAILURE); + } + + LOG.info("Connected client <{}> with login <{}>", connectClientId, username); + } else { + future.channel().pipeline().fireExceptionCaught(future.cause()); + } + } + }); + } + + private boolean isNotProtocolVersion(MqttConnectMessage msg, MqttVersion version) { + return msg.variableHeader().version() != version.protocolLevel(); + } + + private void reauthorizeOnExistingSubscriptions(String clientId, String username) { + if (!m_sessionsStore.contains(clientId)) { + return; + } + final Collection clientSubscriptions = m_sessionsStore.subscriptionStore() + .listClientSubscriptions(clientId); + for (Subscription sub : clientSubscriptions) { + final Topic topicToReauthorize = sub.getTopicFilter(); + final boolean readAuthorized = m_authorizator.canRead(topicToReauthorize, username, clientId); + if (!readAuthorized) { + subscriptions.removeSubscription(topicToReauthorize, clientId); + } + } + } + + private void setupAutoFlusher(Channel channel, int flushIntervalMs) { + try { + channel.pipeline().addAfter( + "idleEventHandler", + "autoFlusher", + new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS)); + } catch (NoSuchElementException nseex) { + // the idleEventHandler is not present on the pipeline + channel.pipeline() + .addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS)); + } + } + + private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) { + return connAck(returnCode, false); + } + + private MqttConnAckMessage connAckWithSessionPresent(MqttConnectReturnCode returnCode) { + return connAck(returnCode, true); + } + + private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) { + MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, + false, 0); + MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent); + return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); + } + + private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) { + // handle user authentication + if (msg.variableHeader().hasUserName()) { + byte[] pwd = null; + if (msg.variableHeader().hasPassword()) { + pwd = msg.payload().password().getBytes(StandardCharsets.UTF_8); + } else if (!this.allowAnonymous) { + LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId); + failedCredentials(channel); + return false; + } + final String login = msg.payload().userName(); + if (!m_authenticator.checkValid(clientId, login, pwd)) { + LOG.error("Authenticator has rejected the MQTT credentials CId={}, username={}", clientId, login); + failedCredentials(channel); + return false; + } + NettyUtils.userName(channel, login); + } else if (!this.allowAnonymous) { + LOG.error("Client didn't supply any credentials and MQTT anonymous mode is disabled. CId={}", clientId); + failedCredentials(channel); + return false; + } + return true; + } + + private MqttConnAckMessage createConnectAck(MqttConnectMessage msg, String clientId) { + MqttConnAckMessage okResp; + ClientSession clientSession = this.sessionsRepository.sessionForClient(clientId); + boolean isSessionAlreadyStored = clientSession != null; + final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession(); + if (!msgCleanSessionFlag && isSessionAlreadyStored) { + okResp = connAckWithSessionPresent(CONNECTION_ACCEPTED); + } else { + okResp = connAck(CONNECTION_ACCEPTED); + } + return okResp; + } + + private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) { + int keepAlive = msg.variableHeader().keepAliveTimeSeconds(); + NettyUtils.keepAlive(channel, keepAlive); + NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession()); + NettyUtils.clientID(channel, clientId); + int idleTime = Math.round(keepAlive * 1.5f); + setIdleTime(channel.pipeline(), idleTime); + + LOG.debug("Connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}", + clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime); + } + + private void failedCredentials(Channel session) { + session.writeAndFlush(connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)) + .addListener(FIRE_EXCEPTION_ON_FAILURE); + LOG.info("Client {} failed to connect with bad username or password.", session); + } + + private void setIdleTime(ChannelPipeline pipeline, int idleTime) { + if (pipeline.names().contains("idleStateHandler")) { + pipeline.remove("idleStateHandler"); + } + pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0)); + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/DisconnectHandler.java b/broker/src/main/java/io/moquette/spi/impl/DisconnectHandler.java new file mode 100644 index 00000000..c9652740 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/DisconnectHandler.java @@ -0,0 +1,133 @@ +package io.moquette.spi.impl; + +import io.moquette.connections.IConnectionsManager; +import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.ClientSession; +import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.netty.channel.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Optional; + +import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; +import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; + +public class DisconnectHandler { + + private static final Logger LOG = LoggerFactory.getLogger(DisconnectHandler.class); + + private final IConnectionsManager connectionDescriptors; + private final SessionsRepository sessionsRepository; + private final ISubscriptionsDirectory subscriptions; + private final BrokerInterceptor m_interceptor; + + public DisconnectHandler(IConnectionsManager connectionDescriptors, SessionsRepository sessionsRepository, + ISubscriptionsDirectory subscriptions, BrokerInterceptor interceptor) { + this.connectionDescriptors = connectionDescriptors; + this.sessionsRepository = sessionsRepository; + this.subscriptions = subscriptions; + this.m_interceptor = interceptor; + } + + public void processDisconnect(Channel channel, String clientID) { + final Optional existingDescriptorOpt = this.connectionDescriptors.lookupDescriptor(clientID); + if (!existingDescriptorOpt.isPresent()) { + // another client with same ID removed the descriptor, we must exit + channel.close().addListener(CLOSE_ON_FAILURE); + return; + } + final ConnectionDescriptor existingDescriptor = existingDescriptorOpt.get(); + + if (existingDescriptor.doesNotUseChannel(channel)) { + // another client saved it's descriptor, exit + LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}", clientID); + existingDescriptor.abort(); + return; + } + + if (!removeSubscriptions(existingDescriptor, clientID)) { + LOG.warn("Unable to remove subscriptions. Closing connection. CId={}", clientID); + existingDescriptor.abort(); + return; + } + + if (!dropStoredMessages(existingDescriptor, clientID)) { + LOG.warn("Unable to drop stored messages. Closing connection. CId={}", clientID); + existingDescriptor.abort(); + return; + } + + if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) { + LOG.warn("Unable to drop will message. Closing connection. CId={}", clientID); + existingDescriptor.abort(); + return; + } + + if (!existingDescriptor.close()) { + LOG.debug("Connection has been closed. CId={}", clientID); + return; + } + + boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); + if (!stillPresent) { + // another descriptor was inserted + LOG.warn("Another descriptor has been inserted. CId={}", clientID); + return; + } + this.sessionsRepository.disconnect(clientID); + + LOG.info("Client <{}> disconnected", clientID); + } + + private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) { + final boolean success = descriptor.assignState(ESTABLISHED, SUBSCRIPTIONS_REMOVED); + if (!success) { + return false; + } + + if (descriptor.cleanSession) { + LOG.trace("Removing saved subscriptions. CId={}", descriptor.clientID); + final ClientSession session = this.sessionsRepository.sessionForClient(clientID); + session.wipeSubscriptions(); + for (Subscription existingSub : session.getSubscriptions()) { + this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientID); + } + LOG.trace("Saved subscriptions have been removed. CId={}", descriptor.clientID); + } + return true; + } + + private boolean dropStoredMessages(ConnectionDescriptor descriptor, String clientID) { + final boolean success = descriptor.assignState(SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED); + if (!success) { + return false; + } + + final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID); + if (clientSession.isCleanSession()) { + clientSession.dropQueue(); + } + return true; + } + + private boolean cleanWillMessageAndNotifyInterceptor(ConnectionDescriptor descriptor, String clientID) { + final boolean success = descriptor.assignState(MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED); + if (!success) { + return false; + } + + LOG.trace("Removing will message. ClientId={}", descriptor.clientID); + // cleanup the will store + final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID); + clientSession.removeWill(); + + String username = descriptor.getUsername(); + m_interceptor.notifyClientDisconnected(clientID, username); + return true; + } + + +} diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 1e0b790c..9b4df487 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -59,6 +59,7 @@ class InternalRepublisher { return; } + LOG.info("Republishing stored publish events. CId={}", clientSession.clientID); EnqueuedMessage pubEvt; while ((pubEvt = clientSession.poll()) != null) { // put in flight zone diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 33d71bbd..d91ca4c8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -19,13 +19,14 @@ package io.moquette.spi.impl; import io.moquette.connections.IConnectionsManager; import io.moquette.interception.InterceptHandler; import io.moquette.interception.messages.InterceptAcknowledgedMessage; -import io.moquette.persistence.MemorySessionStore; import io.moquette.server.ConnectionDescriptor; import io.moquette.server.ConnectionDescriptorStore; -import io.moquette.server.netty.AutoFlushHandler; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.*; +import io.moquette.spi.ClientSession; +import io.moquette.spi.EnqueuedMessage; +import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; @@ -33,18 +34,16 @@ import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; -import io.netty.channel.ChannelPipeline; import io.netty.handler.codec.mqtt.*; -import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeUnit; import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos; @@ -52,7 +51,6 @@ import static io.moquette.spi.impl.Utils.messageId; import static io.moquette.spi.impl.Utils.readBytesAndRewind; import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE; -import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.*; @@ -64,37 +62,6 @@ import static io.netty.handler.codec.mqtt.MqttQoS.*; */ public class ProtocolProcessor { - static final class WillMessage { - - private final String topic; - private final ByteBuffer payload; - private final boolean retained; - private final MqttQoS qos; - - WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) { - this.topic = topic; - this.payload = payload; - this.retained = retained; - this.qos = qos; - } - - public String getTopic() { - return topic; - } - - public ByteBuffer getPayload() { - return payload; - } - - public boolean isRetained() { - return retained; - } - - public MqttQoS getQos() { - return qos; - } - } - private enum SubscriptionState { STORED, VERIFIED } @@ -105,16 +72,12 @@ public class ProtocolProcessor { private ConcurrentMap subscriptionInCourse; private ISubscriptionsDirectory subscriptions; - private boolean allowAnonymous; - private boolean allowZeroByteClientId; - private boolean reauthorizeSubscriptionsOnConnect; private IAuthorizator m_authorizator; private IMessagesStore m_messagesStore; private ISessionsStore m_sessionsStore; - private IAuthenticator m_authenticator; private BrokerInterceptor m_interceptor; private Qos0PublishHandler qos0PublishHandler; @@ -124,8 +87,8 @@ public class ProtocolProcessor { private InternalRepublisher internalRepublisher; SessionsRepository sessionsRepository; - // maps clientID to Will testament, if specified on CONNECT - private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); + private ConnectHandler connectHandler; + private DisconnectHandler disconnectHandler; ProtocolProcessor() { } @@ -170,25 +133,21 @@ public class ProtocolProcessor { boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository, boolean reauthorizeSubscriptionsOnConnect) { - LOG.debug("Initializing MQTT protocol processor..."); + LOG.info("Initializing MQTT protocol processor.."); this.connectionDescriptors = connectionDescriptors; this.subscriptionInCourse = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; - this.allowAnonymous = allowAnonymous; - this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect; - this.allowZeroByteClientId = allowZeroByteClientId; m_authorizator = authorizator; if (LOG.isDebugEnabled()) { LOG.debug("Initial subscriptions tree={}", subscriptions.dumpTree()); } - m_authenticator = authenticator; m_messagesStore = storageService; m_sessionsStore = sessionsStore; this.sessionsRepository = sessionsRepository; - LOG.info("Initializing messages publisher..."); + LOG.debug("Initializing messages publisher..."); final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors); this.messagesPublisher = new MessagesPublisher(connectionDescriptors, messageSender, subscriptions, this.sessionsRepository); @@ -203,240 +162,16 @@ public class ProtocolProcessor { LOG.debug("Initializing internal republisher..."); this.internalRepublisher = new InternalRepublisher(messageSender); + + connectHandler = new ConnectHandler(connectionDescriptors, m_interceptor, sessionsRepository, m_sessionsStore, + authenticator, m_authorizator, subscriptions, + allowAnonymous, allowZeroByteClientId, reauthorizeSubscriptionsOnConnect, internalRepublisher); + disconnectHandler = new DisconnectHandler(connectionDescriptors, sessionsRepository, subscriptions, m_interceptor); + LOG.info("Initialized"); } - public void processConnect(Channel channel, MqttConnectMessage msg) { - MqttConnectPayload payload = msg.payload(); - String clientId = payload.clientIdentifier(); - final String username = payload.userName(); - LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, username); - - if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel() - && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) { - MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION); - - LOG.error("MQTT protocol version is not valid. CId={}", clientId); - channel.writeAndFlush(badProto).addListener(FIRE_EXCEPTION_ON_FAILURE); - channel.close().addListener(CLOSE_ON_FAILURE); - return; - } - - final boolean cleanSession = msg.variableHeader().isCleanSession(); - if (clientId == null || clientId.length() == 0) { - if (!cleanSession || !this.allowZeroByteClientId) { - MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED); - - channel.writeAndFlush(badId).addListener(FIRE_EXCEPTION_ON_FAILURE); - channel.close().addListener(CLOSE_ON_FAILURE); - LOG.error("The MQTT client ID cannot be empty. Username={}", username); - return; - } - - // Generating client id. - clientId = UUID.randomUUID().toString().replace("-", ""); - LOG.info("Client has connected with a server generated identifier. CId={}, username={}", clientId, - username); - } - - if (!login(channel, msg, clientId)) { - channel.close().addListener(CLOSE_ON_FAILURE); - return; - } - - ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession); - final ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor); - if (existing != null) { - LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId); - existing.abort(); - //return; - this.connectionDescriptors.removeConnection(existing); - this.connectionDescriptors.addConnection(descriptor); - } - - initializeKeepAliveTimeout(channel, msg, clientId); - storeWillMessage(msg, clientId); - if (!cleanSession && reauthorizeSubscriptionsOnConnect) { - reauthorizeOnExistingSubscriptions(clientId, username); - } - if (!sendAck(descriptor, msg, clientId)) { - channel.close().addListener(CLOSE_ON_FAILURE); - return; - } - - m_interceptor.notifyClientConnected(msg); - - if (!descriptor.assignState(SENDACK, SESSION_CREATED)) { - channel.close().addListener(CLOSE_ON_FAILURE); - return; - } - final ClientSession clientSession = this.sessionsRepository.createOrLoadClientSession(clientId, cleanSession); - - if (!republish(descriptor, msg, clientSession)) { - channel.close().addListener(CLOSE_ON_FAILURE); - return; - } - - int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */; - setupAutoFlusher(channel, flushIntervalMs); - - final boolean success = descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED); - if (!success) { - channel.close().addListener(CLOSE_ON_FAILURE); - } - - LOG.info("Connected client <{}> with login <{}>", clientId, username); - } - - private void reauthorizeOnExistingSubscriptions(String clientId, String username) { - if (!m_sessionsStore.contains(clientId)) { - return; - } - final Collection clientSubscriptions = m_sessionsStore.subscriptionStore() - .listClientSubscriptions(clientId); - for (Subscription sub : clientSubscriptions) { - final Topic topicToReauthorize = sub.getTopicFilter(); - final boolean readAuthorized = m_authorizator.canRead(topicToReauthorize, username, clientId); - if (!readAuthorized) { - subscriptions.removeSubscription(topicToReauthorize, clientId); - } - } - } - - private void setupAutoFlusher(Channel channel, int flushIntervalMs) { - try { - channel.pipeline().addAfter( - "idleEventHandler", - "autoFlusher", - new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS)); - } catch (NoSuchElementException nseex) { - // the idleEventHandler is not present on the pipeline - channel.pipeline() - .addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS)); - } - } - - private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) { - return connAck(returnCode, false); - } - - private MqttConnAckMessage connAckWithSessionPresent(MqttConnectReturnCode returnCode) { - return connAck(returnCode, true); - } - - private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, - false, 0); - MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent); - return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); - } - - private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) { - // handle user authentication - if (msg.variableHeader().hasUserName()) { - byte[] pwd = null; - if (msg.variableHeader().hasPassword()) { - pwd = msg.payload().password().getBytes(StandardCharsets.UTF_8); - } else if (!this.allowAnonymous) { - LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId); - failedCredentials(channel); - return false; - } - final String login = msg.payload().userName(); - if (!m_authenticator.checkValid(clientId, login, pwd)) { - LOG.error("Authenticator has rejected the MQTT credentials CId={}, username={}", clientId, login); - failedCredentials(channel); - return false; - } - NettyUtils.userName(channel, login); - } else if (!this.allowAnonymous) { - LOG.error("Client didn't supply any credentials and MQTT anonymous mode is disabled. CId={}", clientId); - failedCredentials(channel); - return false; - } - return true; - } - - private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, String clientId) { - LOG.debug("Sending CONNACK. CId={}", clientId); - final boolean success = descriptor.assignState(DISCONNECTED, SENDACK); - if (!success) { - return false; - } - - MqttConnAckMessage okResp; - ClientSession clientSession = this.sessionsRepository.sessionForClient(clientId); - boolean isSessionAlreadyStored = clientSession != null; - final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession(); - if (!msgCleanSessionFlag && isSessionAlreadyStored) { - okResp = connAckWithSessionPresent(CONNECTION_ACCEPTED); - } else { - okResp = connAck(CONNECTION_ACCEPTED); - } - - descriptor.writeAndFlush(okResp); - LOG.debug("CONNACK has been sent. CId={}", clientId); - - if (isSessionAlreadyStored && msgCleanSessionFlag) { - for (Subscription existingSub : clientSession.getSubscriptions()) { - this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientId); - } - } - return true; - } - - private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) { - int keepAlive = msg.variableHeader().keepAliveTimeSeconds(); - NettyUtils.keepAlive(channel, keepAlive); - NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession()); - NettyUtils.clientID(channel, clientId); - int idleTime = Math.round(keepAlive * 1.5f); - setIdleTime(channel.pipeline(), idleTime); - - LOG.debug("Connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}", - clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime); - } - - private void storeWillMessage(MqttConnectMessage msg, String clientId) { - // Handle will flag - if (msg.variableHeader().isWillFlag()) { - MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos()); - LOG.debug("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}", - clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain()); - byte[] willPayload = msg.payload().willMessage().getBytes(StandardCharsets.UTF_8); - ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); - // save the will testament in the clientID store - WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), - willQos); - m_willStore.put(clientId, will); - LOG.debug("MQTT last will and testament has been configured. CId={}", clientId); - } - } - - private boolean republish(ConnectionDescriptor descriptor, MqttConnectMessage msg, ClientSession clientSession) { - final boolean success = descriptor.assignState(SESSION_CREATED, MESSAGES_REPUBLISHED); - if (!success) { - return false; - } - - if (!msg.variableHeader().isCleanSession()) { - // force the republish of stored QoS1 and QoS2 - LOG.info("Republishing stored publish events. CId={}", clientSession.clientID); - this.internalRepublisher.publishStored(clientSession); - } - return true; - } - - private void failedCredentials(Channel session) { - session.writeAndFlush(connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)) - .addListener(FIRE_EXCEPTION_ON_FAILURE); - LOG.info("Client {} failed to connect with bad username or password.", session); - } - - private void setIdleTime(ChannelPipeline pipeline, int idleTime) { - if (pipeline.names().contains("idleStateHandler")) { - pipeline.remove("idleStateHandler"); - } - pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0)); + public void processConnect(Channel channel, MqttConnectMessage connectMessage) { + this.connectHandler.processConnect(channel, connectMessage); } public void processPubAck(Channel channel, MqttPubAckMessage msg) { @@ -476,7 +211,7 @@ public class ProtocolProcessor { final MqttQoS qos = msg.fixedHeader().qosLevel(); final String clientId = NettyUtils.clientID(channel); LOG.info("Processing PUBLISH message. CId={}, topic={}, messageId={}, qos={}", clientId, - msg.variableHeader().topicName(), msg.variableHeader().messageId(), qos); + msg.variableHeader().topicName(), msg.variableHeader().messageId(), qos); switch (qos) { case AT_MOST_ONCE: this.qos0PublishHandler.receivedPublishQos0(channel, msg); @@ -601,100 +336,7 @@ public class ProtocolProcessor { final String clientID = NettyUtils.clientID(channel); LOG.debug("Processing DISCONNECT message. CId={}", clientID); channel.flush(); - final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); - if (existingDescriptor == null) { - // another client with same ID removed the descriptor, we must exit - channel.close().addListener(CLOSE_ON_FAILURE); - return; - } - - if (existingDescriptor.doesNotUseChannel(channel)) { - // another client saved it's descriptor, exit - LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}", clientID); - existingDescriptor.abort(); - return; - } - - if (!removeSubscriptions(existingDescriptor, clientID)) { - LOG.warn("Unable to remove subscriptions. Closing connection. CId={}", clientID); - existingDescriptor.abort(); - return; - } - - if (!dropStoredMessages(existingDescriptor, clientID)) { - LOG.warn("Unable to drop stored messages. Closing connection. CId={}", clientID); - existingDescriptor.abort(); - return; - } - - if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) { - LOG.warn("Unable to drop will message. Closing connection. CId={}", clientID); - existingDescriptor.abort(); - return; - } - - if (!existingDescriptor.close()) { - LOG.debug("Connection has been closed. CId={}", clientID); - return; - } - - boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); - if (!stillPresent) { - // another descriptor was inserted - LOG.warn("Another descriptor has been inserted. CId={}", clientID); - return; - } - this.sessionsRepository.disconnect(clientID); - - LOG.info("Client <{}> disconnected", clientID); - } - - private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor.assignState(ESTABLISHED, SUBSCRIPTIONS_REMOVED); - if (!success) { - return false; - } - - if (descriptor.cleanSession) { - LOG.trace("Removing saved subscriptions. CId={}", descriptor.clientID); - final ClientSession session = this.sessionsRepository.sessionForClient(clientID); - session.wipeSubscriptions(); - for (Subscription existingSub : session.getSubscriptions()) { - this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientID); - } - LOG.trace("Saved subscriptions have been removed. CId={}", descriptor.clientID); - } - return true; - } - - private boolean dropStoredMessages(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor.assignState(SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED); - if (!success) { - return false; - } - - final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID); - if (clientSession.isCleanSession()) { - clientSession.dropQueue(); -// LOG.debug("Removing messages of session. CId={}", descriptor.clientID); -// this.m_sessionsStore.dropQueue(clientID); -// LOG.debug("Messages of the session have been removed. CId={}", descriptor.clientID); - } - return true; - } - - private boolean cleanWillMessageAndNotifyInterceptor(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor.assignState(MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED); - if (!success) { - return false; - } - - LOG.trace("Removing will message. ClientId={}", descriptor.clientID); - // cleanup the will store - m_willStore.remove(clientID); - String username = descriptor.getUsername(); - m_interceptor.notifyClientDisconnected(clientID, username); - return true; + disconnectHandler.processDisconnect(channel, clientID); } public void processConnectionLost(String clientID, Channel channel) { @@ -702,10 +344,11 @@ public class ProtocolProcessor { ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); connectionDescriptors.removeConnection(oldConnDescr); // publish the Will message (if any) for the clientID - if (m_willStore.containsKey(clientID)) { - WillMessage will = m_willStore.get(clientID); + final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID); + WillMessage will = clientSession.willMessage(); + if (will != null) { forwardPublishWill(will, clientID); - m_willStore.remove(clientID); + clientSession.removeWill(); } String username = NettyUtils.userName(channel); diff --git a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java index 3f0c091c..9ebed215 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java +++ b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java @@ -24,7 +24,6 @@ import java.time.LocalDateTime; import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Collection; -import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -50,6 +49,9 @@ public class SessionsRepository { private ISubscriptionsStore subscriptionsStore; private ScheduledExecutorService scheduler; private final ConcurrentMap sessionsCache = new ConcurrentHashMap<>(); + // maps clientID to Will testament, if specified on CONNECT + // maps clientID to Will testament, if specified on CONNECT + private ConcurrentMap willMessages = new ConcurrentHashMap<>(); public SessionsRepository(ISessionsStore sessionsStore, ScheduledExecutorService scheduler) { this.sessions = sessionsStore; @@ -80,9 +82,9 @@ public class SessionsRepository { private ClientSession newClientSessionAndCacheIt(String clientID, boolean cleanSession) { ClientSession session; if (cleanSession) { - session = new TransientSession(clientID); + session = new TransientSession(clientID, this.willMessages); } else { - DurableSession durableSession = new DurableSession(clientID, this.sessions, this.subscriptionsStore); + DurableSession durableSession = new DurableSession(clientID, this.sessions, this.subscriptionsStore, willMessages); durableSession.reloadAllSubscriptionsFromStore(); session = durableSession; } diff --git a/broker/src/main/java/io/moquette/spi/impl/WillMessage.java b/broker/src/main/java/io/moquette/spi/impl/WillMessage.java new file mode 100644 index 00000000..9fe15974 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/WillMessage.java @@ -0,0 +1,36 @@ +package io.moquette.spi.impl; + +import io.netty.handler.codec.mqtt.MqttQoS; + +import java.nio.ByteBuffer; + +public final class WillMessage { + + private final String topic; + private final ByteBuffer payload; + private final boolean retained; + private final MqttQoS qos; + + public WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) { + this.topic = topic; + this.payload = payload; + this.retained = retained; + this.qos = qos; + } + + public String getTopic() { + return topic; + } + + public ByteBuffer getPayload() { + return payload; + } + + public boolean isRetained() { + return retained; + } + + public MqttQoS getQos() { + return qos; + } +} diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java deleted file mode 100644 index 2e0e3ed1..00000000 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * Copyright (c) 2012-2018 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; - -import io.moquette.interception.HazelcastInterceptHandler; -import io.moquette.server.config.IConfig; -import io.moquette.server.config.MemoryConfig; -import org.eclipse.paho.client.mqttv3.IMqttClient; -import org.eclipse.paho.client.mqttv3.MqttClient; -import org.eclipse.paho.client.mqttv3.MqttClientPersistence; -import org.eclipse.paho.client.mqttv3.MqttMessage; -import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; -import org.junit.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.IOException; -import java.security.NoSuchAlgorithmException; -import java.sql.SQLException; -import java.util.Properties; - -import static io.moquette.BrokerConstants.*; -import static java.nio.charset.StandardCharsets.UTF_8; -import static org.junit.Assert.assertEquals; - -public class ServerIntegrationHazelcastHandlerInterceptorTest { - - private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationHazelcastHandlerInterceptorTest.class); - - static MqttClientPersistence s_dataStore; - static MqttClientPersistence s_pubDataStore; - - Server server1883; - Server server1884; - IMqttClient m_listener; - IMqttClient m_publisher; - MessageCollector m_messagesCollector; - IConfig m_config_1883; - IConfig m_config_1884; - - @BeforeClass - public static void beforeTests() throws NoSuchAlgorithmException, SQLException, ClassNotFoundException { - String tmpDir = System.getProperty("java.io.tmpdir"); - s_dataStore = new MqttDefaultFilePersistence(tmpDir); - s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); - } - - protected Server startServer(int port, IConfig m_config) throws IOException { - Server m_server = new Server(); - m_config.setProperty(PORT_PROPERTY_NAME, Integer.toString(port)); - m_server.startServer(m_config); - return m_server; - } - - private void stopServer(Server m_server) { - m_server.stopServer(); - } - - private Properties addHazelCastConf(Properties properties, int port, String hazelcastConfigurationFile) { - properties.put(PORT_PROPERTY_NAME, port); - properties.put(INTERCEPT_HANDLER_PROPERTY_NAME, HazelcastInterceptHandler.class.getCanonicalName()); - properties.put(HAZELCAST_CONFIGURATION, hazelcastConfigurationFile); - return properties; - } - - @Before - public void setUp() throws Exception { - final Properties configProps = addHazelCastConf( - IntegrationUtils.prepareTestClusterProperties(1883), - 1883, - "config/hazelcast.xml"); - m_config_1883 = new MemoryConfig(configProps); - - server1883 = startServer(1883, m_config_1883); - - final Properties configProps1884 = addHazelCastConf( - IntegrationUtils.prepareTestClusterProperties(1884), - 1883, - "config/hazelcast.xml"); - m_config_1884 = new MemoryConfig(configProps1884); - server1884 = startServer(1884, m_config_1884); - - m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); - m_publisher.connect(); - m_listener = new MqttClient("tcp://localhost:1884", "Listener", s_dataStore); - m_messagesCollector = new MessageCollector(); - m_listener.setCallback(m_messagesCollector); - - m_listener.connect(); - } - - @After - public void tearDown() throws Exception { - if (m_listener != null && m_listener.isConnected()) { - m_listener.disconnect(); - } - - if (m_publisher != null && m_publisher.isConnected()) { - m_publisher.disconnect(); - } - - stopServer(server1883); - stopServer(server1884); - } - - @Test - public void checkPublishPassThroughCluster_Qos0() throws Exception { - LOG.info("*** checkPublishPassThroughCluster_Qos0 ***"); - m_listener.subscribe("/topic", 1); - - m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(UTF_8), 0, false); - MqttMessage messageQos0 = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT QoS0", messageQos0.toString()); - assertEquals(0, messageQos0.getQos()); - } - - @Test - public void checkPublishPassThroughCluster_Qos1() throws Exception { - LOG.info("*** checkPublishPassThroughCluster_Qos1 ***"); - m_listener.subscribe("/topic", 1); - - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(UTF_8), 1, false); - MqttMessage messageQos1 = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT QoS1", messageQos1.toString()); - assertEquals(1, messageQos1.getQos()); - } - - @Test - public void checkPublishPassThroughCluster_Qos2() throws Exception { - LOG.info("*** checkPublishPassThroughCluster_Qos2 ***"); - m_listener.subscribe("/topic", 2); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 2, false); - MqttMessage messageQos2 = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT QoS2", messageQos2.toString()); - assertEquals(2, messageQos2.getQos()); - } -} diff --git a/broker/src/test/java/io/moquette/spi/impl/ConnectHandlerTest.java b/broker/src/test/java/io/moquette/spi/impl/ConnectHandlerTest.java new file mode 100644 index 00000000..eac4b2ff --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/ConnectHandlerTest.java @@ -0,0 +1,315 @@ +package io.moquette.spi.impl; + +import io.moquette.persistence.MemoryStorageService; +import io.moquette.server.ConnectionDescriptorStore; +import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.ClientSession; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.impl.security.PermitAllAuthorizator; +import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory; +import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.mqtt.MqttConnectMessage; +import io.netty.handler.codec.mqtt.MqttMessageBuilders; +import io.netty.handler.codec.mqtt.MqttVersion; +import org.junit.Before; +import org.junit.Test; + +import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck; +import static io.moquette.spi.impl.ProtocolProcessorTest.*; +import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; +import static java.util.Collections.singleton; +import static java.util.Collections.singletonMap; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class ConnectHandlerTest { + + EmbeddedChannel session; + MqttMessageBuilders.ConnectBuilder connMsg; + + IMessagesStore m_messagesStore; + ISessionsStore m_sessionStore; + ISubscriptionsDirectory subscriptions; + MockAuthenticator m_mockAuthenticator; + private ConnectHandler sut; + private DisconnectHandler disconnectHandler; + private SessionsRepository sessionsRepository; + + @Before + public void setUp() { + connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true); + + session = new EmbeddedChannel(); + + MemoryStorageService memStorage = new MemoryStorageService(null, null); + m_messagesStore = memStorage.messagesStore(); + m_sessionStore = memStorage.sessionsStore(); + m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); + + subscriptions = new CTrieSubscriptionDirectory(); + sessionsRepository = new SessionsRepository(m_sessionStore, null); + subscriptions.init(sessionsRepository); + + final ConnectionDescriptorStore connectionsRegistry = new ConnectionDescriptorStore(); + sut = new ConnectHandler(connectionsRegistry, NO_OBSERVERS_INTERCEPTOR, sessionsRepository, m_sessionStore, m_mockAuthenticator, + new PermitAllAuthorizator(), subscriptions, true, true, false, null); + + disconnectHandler = new DisconnectHandler(connectionsRegistry, sessionsRepository, + subscriptions, NO_OBSERVERS_INTERCEPTOR); + } + + @Test + public void testZeroByteClientIdWithCleanSession() { + // Allow zero byte client ids + sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator, + new PermitAllAuthorizator(), subscriptions, true, true, false, null); + + // Connect message with clean session set to true and client id is null. + MqttConnectMessage msg = connMsg.clientId(null) + .protocolVersion(MqttVersion.MQTT_3_1_1) + .cleanSession(true) + .build(); + + sut.processConnect(session, msg); + assertEqualsConnAck("Connection MUST be accepted, unique clientid MUST be generated and clean session to true", + CONNECTION_ACCEPTED, session.readOutbound()); + assertTrue("Connection should be valid and open,", session.isOpen()); + } + + @Test + public void invalidAuthentication() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER + "_fake") + .password(TEST_PWD) + .build(); + + // Exercise + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound()); + assertFalse("Connection should be closed by the broker.", session.isOpen()); + } + + @Test + public void testConnect_badClientID() { + connMsg.clientId("extremely_long_clientID_greater_than_23").build(); + + // Exercise + sut.processConnect(session, connMsg.clientId("extremely_long_clientID_greater_than_23").build()); + + // Verify + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + } + + @Test + public void testWill() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true) + .willTopic("topic").willMessage("Topic message").build(); + + // Exercise + // m_handler.setMessaging(mockedMessaging); + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", session.isOpen()); + // TODO verify the call + /* + * verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), + * any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class)); + */ + } + + @Test + public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() { + // Connect a client1 + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER) + .password(TEST_PWD) + .build(); + sut.processConnect(session, msg); + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + + // create another connect same clientID but with bad credentials + MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect() + .protocolVersion(MqttVersion.MQTT_3_1) + .clientId(FAKE_CLIENT_ID) + .username(ProtocolProcessorTest.EVIL_TEST_USER) + .password(ProtocolProcessorTest.EVIL_TEST_PWD) + .build(); + + EmbeddedChannel evilSession = new EmbeddedChannel(); + + // Exercise + sut.processConnect(evilSession, evilClientConnMsg); + + // Verify + // the evil client gets a not auth notification + assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, evilSession.readOutbound()); + // the good client remains connected + assertTrue(session.isOpen()); + assertFalse(evilSession.isOpen()); + } + + @Test + public void acceptAnonymousClient() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build(); + + // Exercise + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", session.isOpen()); + } + + @Test + public void validAuthentication() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER).password(TEST_PWD).build(); + + // Exercise + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", session.isOpen()); + } + + @Test + public void noPasswdAuthentication() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER) + .build(); + + // Exercise + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound()); + assertFalse("Connection should be closed by the broker.", session.isOpen()); + } + + @Test + public void prohibitAnonymousClient() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build(); + reinitProcessorProhibitingAnonymousClients(); + + // Exercise + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound()); + assertFalse("Connection should be closed by the broker.", session.isOpen()); + } + + protected void reinitProcessorProhibitingAnonymousClients() { + sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator, + new PermitAllAuthorizator(), subscriptions, false, true, false, null); + } + + @Test + public void prohibitAnonymousClient_providingUsername() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER + "_fake") + .build(); + reinitProcessorProhibitingAnonymousClients(); + + // Exercise + sut.processConnect(session, msg); + + // Verify + assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, session.readOutbound()); + assertFalse("Connection should be closed by the broker.", session.isOpen()); + } + + @Test + public void testZeroByteClientIdNotAllowed() { + sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator, + new PermitAllAuthorizator(), subscriptions, false, false, false, null); + + // Connect message with clean session set to true and client id is null. + MqttConnectMessage msg = connMsg.clientId(null) + .protocolVersion(MqttVersion.MQTT_3_1_1) + .cleanSession(true) + .build(); + + sut.processConnect(session, msg); + assertEqualsConnAck("Zero byte client identifiers are not allowed", + CONNECTION_REFUSED_IDENTIFIER_REJECTED, session.readOutbound()); + + assertFalse("Connection should closed.", session.isOpen()); + } + + @Test + public void testZeroByteClientIdWithoutCleanSession() { + // Allow zero byte client ids + reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients(); + + // Connect message without clean session set to true but client id is still null + MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1) + .build(); + + sut.processConnect(session, msg); + assertEqualsConnAck( + "Identifier should be rejected due to having clean session set to false.", + CONNECTION_REFUSED_IDENTIFIER_REJECTED, + session.readOutbound()); + + assertFalse("Connection should be closed by the broker.", session.isOpen()); + } + + protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() { + sut = new ConnectHandler(new ConnectionDescriptorStore(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), m_sessionStore, m_mockAuthenticator, + new PermitAllAuthorizator(), subscriptions, true, true, false, null); + } + + @Test + public void testConnAckContainsSessionPresentFlag() { + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .protocolVersion(MqttVersion.MQTT_3_1_1).build(); + NettyUtils.clientID(session, FAKE_CLIENT_ID); + NettyUtils.cleanSession(session, false); + + // Connect a first time + sut.processConnect(session, msg); + // disconnect + disconnectHandler.processDisconnect(session, FAKE_CLIENT_ID); + + // Exercise, reconnect + EmbeddedChannel firstReceiverSession = new EmbeddedChannel(); + sut.processConnect(firstReceiverSession, msg); + + // Verify + assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen()); + } + + @Test + public void connectWithCleanSessionUpdateClientSession() { + // first connect with clean session true + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build(); + sut.processConnect(session, msg); + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + disconnectHandler.processDisconnect(session, FAKE_CLIENT_ID); + assertFalse(session.isOpen()); + + // second connect with clean session false + session = new EmbeddedChannel(); + MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID) + .protocolVersion(MqttVersion.MQTT_3_1).build(); + + sut.processConnect(session, secondConnMsg); + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + + // Verify client session is clean false + ClientSession session = sessionsRepository.sessionForClient(FAKE_CLIENT_ID); + assertFalse(session.isCleanSession()); + + // Verify + // assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + } +} 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 8cdaf2a8..ca2e262f 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 @@ -18,7 +18,6 @@ package io.moquette.spi.impl; import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; @@ -34,7 +33,7 @@ import org.junit.Test; import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck; import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck; import static io.moquette.spi.impl.ProtocolProcessorTest.*; -import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; +import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED; import static java.util.Collections.singleton; import static java.util.Collections.singletonMap; import static org.junit.Assert.assertFalse; @@ -42,322 +41,61 @@ import static org.junit.Assert.assertTrue; public class ProtocolProcessor_CONNECT_Test { - EmbeddedChannel m_session; - MqttMessageBuilders.ConnectBuilder connMsg; - ProtocolProcessor m_processor; - - IMessagesStore m_messagesStore; - ISessionsStore m_sessionStore; - ISubscriptionsDirectory subscriptions; - MockAuthenticator m_mockAuthenticator; + private EmbeddedChannel session; + private ProtocolProcessor processor; + private ISessionsStore sessionStore; @Before - public void setUp() throws InterruptedException { - connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true); - - m_session = new EmbeddedChannel(); + public void setUp() { + session = new EmbeddedChannel(); MemoryStorageService memStorage = new MemoryStorageService(null, null); - m_messagesStore = memStorage.messagesStore(); - m_sessionStore = memStorage.sessionsStore(); - // m_messagesStore.initStore(); + IMessagesStore m_messagesStore = memStorage.messagesStore(); + sessionStore = memStorage.sessionsStore(); - m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); + MockAuthenticator mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); - subscriptions = new CTrieSubscriptionDirectory(); - SessionsRepository sessionsRepository = new SessionsRepository(m_sessionStore, null); + ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); + SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); subscriptions.init(sessionsRepository); - m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, + processor = new ProtocolProcessor(); + processor.init(subscriptions, m_messagesStore, sessionStore, mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, - new SessionsRepository(this.m_sessionStore, null), false); + new SessionsRepository(this.sessionStore, null), false); } @Test - public void testConnect_badClientID() { - connMsg.clientId("extremely_long_clientID_greater_than_23").build(); - - // Exercise - m_processor.processConnect(m_session, connMsg.clientId("extremely_long_clientID_greater_than_23").build()); - - // Verify - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - } - - @Test - public void testWill() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true) - .willTopic("topic").willMessage("Topic message").build(); - - // Exercise - // m_handler.setMessaging(mockedMessaging); - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); - // TODO verify the call - /* - * verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), - * any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class)); - */ - } - - @Test - public void validAuthentication() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) - .username(TEST_USER).password(TEST_PWD).build(); - - // Exercise - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); - } - - @Test - public void noPasswdAuthentication() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) - .username(TEST_USER) - .build(); - - // Exercise - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound()); - assertFalse("Connection should be closed by the broker.", m_session.isOpen()); - } - - @Test - public void invalidAuthentication() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) - .username(TEST_USER + "_fake") - .password(TEST_PWD) - .build(); - - // Exercise - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound()); - assertFalse("Connection should be closed by the broker.", m_session.isOpen()); - } - - @Test - public void prohibitAnonymousClient() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build(); - reinitProcessorProhibitingAnonymousClients(); - - // Exercise - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound()); - assertFalse("Connection should be closed by the broker.", m_session.isOpen()); - } - - protected void reinitProcessorProhibitingAnonymousClients() { - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), false); - } - - @Test - public void prohibitAnonymousClient_providingUsername() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) - .username(TEST_USER + "_fake") - .build(); - reinitProcessorProhibitingAnonymousClients(); - - // Exercise - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, m_session.readOutbound()); - assertFalse("Connection should be closed by the broker.", m_session.isOpen()); - } - - @Test - public void acceptAnonymousClient() { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build(); - - // Exercise - m_processor.processConnect(m_session, msg); - - // Verify - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); - } - - @Test - public void connectWithCleanSessionUpdateClientSession() throws InterruptedException { - // first connect with clean session true - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build(); - m_processor.processConnect(m_session, msg); - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - m_processor.processDisconnect(m_session); - assertFalse(m_session.isOpen()); - - // second connect with clean session false - m_session = new EmbeddedChannel(); - MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID) - .protocolVersion(MqttVersion.MQTT_3_1).build(); - - m_processor.processConnect(m_session, secondConnMsg); - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - - // Verify client session is clean false - ClientSession session = m_processor.sessionsRepository.sessionForClient(FAKE_CLIENT_ID); - assertFalse(session.isCleanSession()); - - // Verify - // assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - } - - @Test - public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() { - // Connect a client1 - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) - .username(TEST_USER) - .password(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 = MqttMessageBuilders.connect() - .protocolVersion(MqttVersion.MQTT_3_1) - .clientId(FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.EVIL_TEST_USER) - .password(ProtocolProcessorTest.EVIL_TEST_PWD) - .build(); - - EmbeddedChannel evilSession = new EmbeddedChannel(); - - // Exercise - m_processor.processConnect(evilSession, evilClientConnMsg); - - // Verify - // the evil client gets a not auth notification - assertEqualsConnAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, evilSession.readOutbound()); - // the good client remains connected - assertTrue(m_session.isOpen()); - assertFalse(evilSession.isOpen()); - } - - @Test - public void testConnAckContainsSessionPresentFlag() throws InterruptedException { - MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) - .protocolVersion(MqttVersion.MQTT_3_1_1).build(); - NettyUtils.clientID(m_session, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_session, false); - - // Connect a first time - m_processor.processConnect(m_session, msg); - // disconnect - m_processor.processDisconnect(m_session); - - // Exercise, reconnect - EmbeddedChannel firstReceiverSession = new EmbeddedChannel(); - m_processor.processConnect(firstReceiverSession, msg); - - // Verify - assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound()); - assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen()); - } - - @Test - public void testMultipleReconnection() throws InterruptedException { + public void testMultipleReconnection() { // connect with clean a false and subscribe to a topic MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(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()); + processor.processConnect(session, msg); + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", session.isOpen()); // subscribe MqttSubscribeMessage subscribeMsg = MqttMessageBuilders.subscribe() .addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.NEWS_TOPIC).messageId(10).build(); - NettyUtils.clientID(m_session, FAKE_CLIENT_ID); - NettyUtils.cleanSession(m_session, false); - m_processor.processSubscribe(m_session, subscribeMsg); - Subscription expectedSubscription = new Subscription( - FAKE_CLIENT_ID, - new Topic(ProtocolProcessorTest.NEWS_TOPIC), - MqttQoS.AT_MOST_ONCE); - verifySubscriptionExists(m_sessionStore, expectedSubscription); - assertEqualsSubAck(m_session.readOutbound()); + NettyUtils.clientID(session, FAKE_CLIENT_ID); + NettyUtils.cleanSession(session, false); + processor.processSubscribe(session, subscribeMsg); + final Topic newTopic = new Topic(ProtocolProcessorTest.NEWS_TOPIC); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, newTopic, MqttQoS.AT_MOST_ONCE); + verifySubscriptionExists(sessionStore, expectedSubscription); + assertEqualsSubAck(session.readOutbound()); // disconnect - m_processor.processDisconnect(m_session); - assertFalse(m_session.isOpen()); + processor.processDisconnect(session); + assertFalse(session.isOpen()); // reconnect clean session a false - m_session = new EmbeddedChannel(); - m_processor.processConnect(m_session, msg); - assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); - assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); + session = new EmbeddedChannel(); + processor.processConnect(session, msg); + assertEqualsConnAck(CONNECTION_ACCEPTED, session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", session.isOpen()); // verify that the first subscription is still preserved - verifySubscriptionExists(m_sessionStore, expectedSubscription); - } - - @Test - public void testZeroByteClientIdWithoutCleanSession() { - // Allow zero byte client ids - m_processor = new ProtocolProcessor(); - reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients(); - - // Connect message without clean session set to true but client id is still null - MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1) - .build(); - - m_processor.processConnect(m_session, msg); - assertEqualsConnAck( - "Identifier should be rejected due to having clean session set to false.", - CONNECTION_REFUSED_IDENTIFIER_REJECTED, - m_session.readOutbound()); - - assertFalse("Connection should be closed by the broker.", m_session.isOpen()); - } - - protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() { - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), false); - } - - @Test - public void testZeroByteClientIdWithCleanSession() { - // Allow zero byte client ids - m_processor = new ProtocolProcessor(); - reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients(); - - // Connect message with clean session set to true and client id is null. - MqttConnectMessage msg = connMsg.clientId(null) - .protocolVersion(MqttVersion.MQTT_3_1_1) - .cleanSession(true) - .build(); - - m_processor.processConnect(m_session, msg); - assertEqualsConnAck("Connection MUST be accepted, unique clientid MUST be generated and clean session to true", - CONNECTION_ACCEPTED, m_session.readOutbound()); - assertTrue("Connection should be valid and open,", m_session.isOpen()); - } - - @Test - public void testZeroByteClientIdNotAllowed() { - // Connect message with clean session set to true and client id is null. - MqttConnectMessage msg = connMsg.clientId(null) - .protocolVersion(MqttVersion.MQTT_3_1_1) - .cleanSession(true) - .build(); - - m_processor.processConnect(m_session, msg); - assertEqualsConnAck("Zero byte client identifiers are not allowed", - CONNECTION_REFUSED_IDENTIFIER_REJECTED, m_session.readOutbound()); - - assertFalse("Connection should closed.", m_session.isOpen()); + verifySubscriptionExists(sessionStore, expectedSubscription); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java index 3b61ac2b..bd84b231 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java @@ -32,7 +32,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void testCleanSession_maintainClientSubscriptions() throws InterruptedException { + public void testCleanSession_maintainClientSubscriptions() { connectAsClient(this.m_channel, "TestClient"); subscribe("/topic", AT_MOST_ONCE); disconnect(); @@ -48,7 +48,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc @Ignore("At the end it's equal to testCleanSession_maintainClientSubscriptions") @Test - public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws InterruptedException { + public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() { connectAsClient("TestClient"); subscribe("/topic", AT_MOST_ONCE); disconnect(); @@ -68,7 +68,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc * previous subscription */ @Test - public void testCleanSession_correctlyClientSubscriptions() throws InterruptedException { + public void testCleanSession_correctlyClientSubscriptions() { connectAsClient("TestClient"); subscribe("/topic", AT_MOST_ONCE); disconnect(); @@ -109,7 +109,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException { + public void testUnsubscribe_do_not_notify_anymore_new_session() { connect(); subscribe("/topic", AT_MOST_ONCE); publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false); @@ -125,7 +125,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void testPublishWithQoS1() throws InterruptedException { + public void testPublishWithQoS1() { connect(); subscribe("/topic", AT_LEAST_ONCE); publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, false); @@ -148,21 +148,21 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc verifyPublishIsReceived(); } - protected void publishFromAnotherClient(String topic, MqttQoS qos) { + private void publishFromAnotherClient(String topic, MqttQoS qos) { EmbeddedChannel anotherChannel = new EmbeddedChannel(); connectAsClient(anotherChannel, "AnotherClient"); publishToAs(anotherChannel, "AnotherClient", topic, qos, 67, false); disconnect(anotherChannel); } - protected void publishFromAnotherClient(String topic, String payload, MqttQoS qos) { + private void publishFromAnotherClient(String topic, String payload, MqttQoS qos) { EmbeddedChannel anotherChannel = new EmbeddedChannel(); connectAsClient(anotherChannel, "AnotherClient"); publishToAs(anotherChannel, "AnotherClient", topic, payload, qos, 68, false); disconnect(anotherChannel); } - protected void publishQoS2FromAnotherClient(String topic) { + private void publishQoS2FromAnotherClient(String topic) { EmbeddedChannel anotherChannel = new EmbeddedChannel(); connectAsClient(anotherChannel, "AnotherClient"); publishQoS2ToAs(anotherChannel, "AnotherClient", topic, 67, false); @@ -187,7 +187,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void testPublishWithQoS2() throws InterruptedException { + public void testPublishWithQoS2() { connect(); subscribe("/topic", EXACTLY_ONCE); disconnect(); @@ -201,8 +201,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() - throws InterruptedException { + public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() { connect(); subscribe("/topic", AT_LEAST_ONCE); disconnect(); @@ -223,7 +222,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws InterruptedException { + public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() { connect(); subscribe("/topic", AT_MOST_ONCE); publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, true); @@ -240,7 +239,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws InterruptedException { + public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() { EmbeddedChannel publisher = new EmbeddedChannel(); connectAsClient(publisher, "Publisher"); @@ -261,7 +260,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc } @Test - public void checkReplayStoredPublish_forNoCleanSession_qos1() throws InterruptedException { + public void checkReplayStoredPublish_forNoCleanSession_qos1() { EmbeddedChannel publisher = new EmbeddedChannel(); connectAsClient(publisher, "Publisher"); @@ -283,7 +282,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc * "hello2" */ @Test - public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws InterruptedException { + public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() { connect(); subscribe("topic", EXACTLY_ONCE); disconnect();