diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 992673fa..a280322f 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -120,4 +120,15 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { } ctx.fireChannelWritabilityChanged(); } + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { + if (evt instanceof ProtocolProcessor.ForceDisconnectEvent) { + ProtocolProcessor.ForceDisconnectEvent disconnectEvent = (ProtocolProcessor.ForceDisconnectEvent) evt; + m_processor.forceCloseConnection(ctx.channel(), disconnectEvent.newChannel, disconnectEvent.msg); + } else if (evt instanceof ProtocolProcessor.CompleteConnect) { + ProtocolProcessor.CompleteConnect completeConnectEvent = (ProtocolProcessor.CompleteConnect) evt; + m_processor.completeConnect(ctx.channel(), completeConnectEvent.msg); + } + } } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java index cab996b1..c8280861 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java @@ -16,6 +16,7 @@ package io.moquette.server.netty; import io.moquette.server.Constants; +import io.moquette.spi.impl.ProtocolProcessor; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.util.Attribute; @@ -30,12 +31,14 @@ public class NettyUtils { public static final String ATTR_USERNAME = "username"; public static final String ATTR_SESSION_STOLEN = "sessionStolen"; + public static final String ATTR_CHANNEL_STATUS = "channelStatus"; private static final AttributeKey ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE); private static final AttributeKey ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION); private static final AttributeKey ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID); private static final AttributeKey ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME); private static final AttributeKey ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN); + private static final AttributeKey ATTR_KEY_CHANNEL_STATUS = AttributeKey.valueOf(ATTR_CHANNEL_STATUS); public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey key) { Attribute attr = ctx.channel().attr(key); @@ -77,4 +80,12 @@ public class NettyUtils { public static Boolean sessionStolen(Channel channel) { return (Boolean) channel.attr(NettyUtils.ATTR_KEY_SESSION_STOLEN).get(); } + + public static void channelStatus(Channel channel, ProtocolProcessor.ConnectState newStatus) { + channel.attr(NettyUtils.ATTR_KEY_CHANNEL_STATUS).set(newStatus); + } + + public static ProtocolProcessor.ConnectState channelStatus(Channel channel) { + return (ProtocolProcessor.ConnectState) channel.attr(NettyUtils.ATTR_KEY_CHANNEL_STATUS).get(); + } } diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 91f5c215..90613f58 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -61,8 +61,6 @@ public class ClientSession { private volatile boolean cleanSession; - private boolean active = false; - private BlockingQueue m_queueToPublish = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE); public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore, @@ -133,16 +131,16 @@ public class ClientSession { cleanSession(); } - //deactivate the session - deactivate(); } public void cleanSession() { LOG.info("cleaning old saved subscriptions for client <{}>", this.clientID); m_sessionsStore.wipeSubscriptions(this.clientID); + LOG.debug("Wiped subscriptions for client <{}>", this.clientID); //remove also the messages stored of type QoS1/2 messagesStore.dropMessagesInSession(this.clientID); + LOG.debug("Removed messages in session for client <{}>", this.clientID); } public boolean isCleanSession() { @@ -154,18 +152,6 @@ public class ClientSession { this.m_sessionsStore.updateCleanStatus(this.clientID, cleanSession); } - public void activate() { - this.active = true; - } - - public void deactivate() { - this.active = false; - } - - public boolean isActive() { - return this.active; - } - public int nextPacketId() { return this.m_sessionsStore.nextPacketID(this.clientID); } 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 c5e89f58..68cb77ac 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -95,8 +95,30 @@ public class ProtocolProcessor { } + public class ForceDisconnectEvent { + public final Channel newChannel; + public final ConnectMessage msg; + + public ForceDisconnectEvent(Channel newChannel, ConnectMessage msg) { + this.newChannel = newChannel; + this.msg = msg; + } + } + + public class CompleteConnect { + public final ConnectMessage msg; + + public CompleteConnect(ConnectMessage msg) { + this.msg = msg; + } + } + private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); + public enum ConnectState { + CONNECTED, DISCONNECTING, DISCONNECTED, DROP_BY_CONNECT + } + protected ConcurrentMap m_clientIDs; private SubscriptionsStore subscriptions; private boolean allowAnonymous; @@ -111,8 +133,9 @@ public class ProtocolProcessor { //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); - ProtocolProcessor() {} + private final ConcurrentMap connectionsStatus = new ConcurrentHashMap<>(); + ProtocolProcessor() {} public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, @@ -213,17 +236,35 @@ public class ProtocolProcessor { } //if an old client with the same ID already exists close its session. - if (m_clientIDs.containsKey(msg.getClientID())) { + if (m_clientIDs.containsKey(msg.getClientID()) /*|| (!otherDisconnecting && existingClientSameId)*/) { LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); //clean the subscriptions if the old used a cleanSession = true - Channel oldChannel = m_clientIDs.get(msg.getClientID()).channel; - ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID()); - oldClientSession.disconnect(); - NettyUtils.sessionStolen(oldChannel, true); - oldChannel.close(); - LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); + try { + Channel oldChannel = m_clientIDs.get(msg.getClientID()).channel; + oldChannel.pipeline().fireUserEventTriggered(new ForceDisconnectEvent(channel, msg)); + LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); + } catch (NullPointerException npex) { + //OK this could happen with concurrent DISCONNECT and CONNECT on same clientID + channel.pipeline().fireUserEventTriggered(new CompleteConnect(msg)); + } + } else { + completeConnect(channel, msg); } + } + + public void forceCloseConnection(Channel oldChannel, Channel newChannel, ConnectMessage msg) { + String clientID = NettyUtils.clientID(oldChannel); + ClientSession oldClientSession = m_sessionsStore.sessionForClient(clientID); + oldClientSession.disconnect(); + NettyUtils.sessionStolen(oldChannel, true); + NettyUtils.channelStatus(oldChannel, ConnectState.DROP_BY_CONNECT); + oldChannel.close(); + + newChannel.pipeline().fireUserEventTriggered(new CompleteConnect(msg)); + } + + public void completeConnect(Channel channel, ConnectMessage msg) { ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), channel, msg.isCleanSession()); m_clientIDs.put(msg.getClientID(), connDescr); @@ -268,7 +309,8 @@ public class ProtocolProcessor { LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession()); } - clientSession.activate(); + this.connectionsStatus.put(msg.getClientID(), ConnectState.CONNECTED); + NettyUtils.channelStatus(channel, ConnectState.CONNECTED); if (msg.isCleanSession()) { clientSession.cleanSession(); } @@ -352,7 +394,7 @@ public class ProtocolProcessor { } String clientID = targetSession.clientID; if (m_clientIDs.containsKey(clientID)) { - targetSession.activate(); + this.connectionsStatus.put(clientID, ConnectState.CONNECTED); } } @@ -509,21 +551,23 @@ public class ProtocolProcessor { ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); verifyToActivate(targetSession); + boolean targetIsActive = this.connectionsStatus.get(sub.getClientId()) == ConnectState.CONNECTED; + LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", - sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive()); + sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); ByteBuffer message = origMessage.duplicate(); - if (qos == AbstractMessage.QOSType.MOST_ONE && targetSession.isActive()) { + if (qos == AbstractMessage.QOSType.MOST_ONE && targetIsActive) { //QoS 0 directSend(targetSession, topic, qos, message, false, null); } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it - if (!targetSession.isCleanSession() && !targetSession.isActive()) { + if (!targetSession.isCleanSession() && !targetIsActive) { //store the message in targetSession queue to deliver targetSession.enqueueToDeliver(guid); } else { //publish - if (targetSession.isActive()) { + if (targetIsActive) { int messageId = targetSession.nextPacketId(); targetSession.inFlightAckWaiting(guid, messageId); directSend(targetSession, topic, qos, message, false, messageId); @@ -671,20 +715,38 @@ public class ProtocolProcessor { public void processDisconnect(Channel channel) throws InterruptedException { channel.flush(); + ConnectState status = NettyUtils.channelStatus(channel); + if (status != null && status == ConnectState.DROP_BY_CONNECT) { + LOG.info("DISCONNECT drop by other connect"); + return; + } String clientID = NettyUtils.clientID(channel); boolean cleanSession = NettyUtils.cleanSession(channel); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); + boolean notNewConnection = (status != null && status == ConnectState.CONNECTED); + if (!notNewConnection) { + //while disconnect someone else connected + return; + } + NettyUtils.channelStatus(channel, ConnectState.DISCONNECTING); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); clientSession.disconnect(); - - m_clientIDs.remove(clientID); channel.close(); + status = NettyUtils.channelStatus(channel); + boolean stillDisconnecting = (status != null && status == ConnectState.DISCONNECTING); + if (!stillDisconnecting) { + //someone else changed the state (another connect with same client ID) + return; + } + m_clientIDs.remove(clientID); //cleanup the will store m_willStore.remove(clientID); String username = NettyUtils.userName(channel); m_interceptor.notifyClientDisconnected(clientID, username); +// this.connectionsStatus.remove(clientID, ConnectState.DISCONNECTED); + this.connectionsStatus.remove(clientID); LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession); } @@ -695,7 +757,7 @@ public class ProtocolProcessor { if (sessionStolen) { //de-activate the subscriptions for this ClientID ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - clientSession.deactivate(); + //TODO it's needed? clientSession.deactivate(); LOG.info("Lost connection with client <{}>", clientID); } //publish the Will message (if any) for the clientID diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 15707b74..c79a122c 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -382,7 +382,7 @@ public class ProtocolProcessorTest { @Test public void publishNoPublishToInactiveSession() { //create an inactive session for Subscriber - m_sessionStore.createNewSession("Subscriber", false).deactivate(); + m_sessionStore.createNewSession("Subscriber", false); SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE); @@ -455,8 +455,8 @@ public class ProtocolProcessorTest { memStore.initStore(); IMessagesStore memoryMessageStore = memStore.messagesStore(); ISessionsStore sessionsStore = memStore.sessionsStore(); - sessionsStore.createNewSession("Sub A", false).activate(); - sessionsStore.createNewSession("Sub B", false).activate(); + sessionsStore.createNewSession("Sub A", false); + sessionsStore.createNewSession("Sub B", false); Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE); Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE); diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 16457d76..d29754f8 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -379,7 +379,7 @@ public class SubscriptionsStoreTest { ISessionsStore sessionsStore = memStore.sessionsStore(); aStore.init(sessionsStore); //subscribe a not active clientID1 to /topic - sessionsStore.createNewSession("FAKE_CLI_ID_1", true).deactivate(); + sessionsStore.createNewSession("FAKE_CLI_ID_1", true); Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashSub); aStore.add(slashSub.asClientTopicCouple());