From 38746ad0dd7edfe4d92f835c7e046a4be6ffc5d8 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 16 Nov 2016 22:18:30 +0100 Subject: [PATCH] Moved existing CONNECT processing to an FSM --- .../moquette/server/ConnectionDescriptor.java | 26 +- .../server/netty/NettyMQTTHandler.java | 10 - .../moquette/spi/impl/ProtocolProcessor.java | 252 ++++++++++-------- .../spi/impl/ProtocolProcessorTest.java | 4 +- 4 files changed, 174 insertions(+), 118 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 36c16768..9650e808 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -16,6 +16,10 @@ package io.moquette.server; import io.netty.channel.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.atomic.AtomicReference; /** * Value object to maintain the information of single connection, like ClientID, Channel, @@ -25,10 +29,17 @@ import io.netty.channel.Channel; * @author andrea */ public class ConnectionDescriptor { + + private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptor.class); + + public enum ConnectionState { + DISCONNECTED, SENDACK, SESSION_CREATED, MESSAGES_REPUBLISHED, ESTABLISHED; + } public final String clientID; public final Channel channel; public final boolean cleanSession; + private final AtomicReference channelState = new AtomicReference<>(ConnectionState.DISCONNECTED); public ConnectionDescriptor(String clientID, Channel session, boolean cleanSession) { this.clientID = clientID; @@ -36,9 +47,22 @@ public class ConnectionDescriptor { this.cleanSession = cleanSession; } + public void abort() { + //TODO fire and event to the channel or simply close it? + LOG.error("TODO send an abort event to the channel"); + this.channel.close(); + } + + public boolean assignState(ConnectionState expected, ConnectionState newState) { + return channelState.compareAndSet(expected, newState); + } + @Override public String toString() { - return "ConnectionDescriptor{" + "clientID=" + clientID + ", cleanSession=" + cleanSession + '}'; + return "ConnectionDescriptor{" + "clientID=" + clientID + + ", cleanSession=" + cleanSession + + ", state=" + channelState.get() + + '}'; } @Override 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 a280322f..ae556e69 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -121,14 +121,4 @@ 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/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 68cb77ac..ad924c31 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit; import io.moquette.interception.InterceptHandler; import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.ConnectionDescriptor.ConnectionState; import io.moquette.server.netty.AutoFlushHandler; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.*; @@ -95,31 +96,13 @@ 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; + protected ConcurrentMap connectionDescriptors; private SubscriptionsStore subscriptions; private boolean allowAnonymous; private boolean allowZeroByteClientId; @@ -133,6 +116,7 @@ public class ProtocolProcessor { //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); + @Deprecated private final ConcurrentMap connectionsStatus = new ConcurrentHashMap<>(); ProtocolProcessor() {} @@ -176,7 +160,7 @@ public class ProtocolProcessor { IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { - this.m_clientIDs = new ConcurrentHashMap<>(); + this.connectionDescriptors = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; this.allowAnonymous = allowAnonymous; @@ -191,6 +175,7 @@ public class ProtocolProcessor { public void processConnect(Channel channel, ConnectMessage msg) { LOG.debug("CONNECT for client <{}>", msg.getClientID()); + if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); @@ -215,6 +200,51 @@ public class ProtocolProcessor { LOG.info("Client connected with server generated identifier: {}", randomIdentifier); } + if (!login(channel, msg)) { + channel.close(); + return; + } + + final String clientID = msg.getClientID(); + ConnectionDescriptor descriptor = new ConnectionDescriptor(clientID, channel, msg.isCleanSession()); + ConnectionDescriptor existing = this.connectionDescriptors.putIfAbsent(clientID, descriptor); + if (existing != null) { + LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); + existing.abort(); + return; + } + + initializeKeepAliveTimeout(channel, msg); + + storeWillMessage(msg); + + if (!sendAck(descriptor, msg)) { + channel.close(); + return; + } + + m_interceptor.notifyClientConnected(msg); + + final ClientSession clientSession = createOrLoadClientSession(descriptor, msg); + if (clientSession == null) { + channel.close(); + return; + } + + if (!republish(descriptor, msg, clientSession)) { + channel.close(); + return; + } + final boolean success = descriptor.assignState(ConnectionState.MESSAGES_REPUBLISHED, ConnectionState.ESTABLISHED); + if (!success) { + channel.close(); + } else { + LOG.info("Connection established"); + } + LOG.info("CONNECT processed"); + } + + private boolean login(Channel channel, ConnectMessage msg) { //handle user authentication if (msg.isUserFlag()) { byte[] pwd = null; @@ -222,75 +252,30 @@ public class ProtocolProcessor { pwd = msg.getPassword(); } else if (!this.allowAnonymous) { failedCredentials(channel); - return; + return false; } if (!m_authenticator.checkValid(msg.getClientID(), msg.getUsername(), pwd)) { failedCredentials(channel); - channel.close(); - return; + return false; } NettyUtils.userName(channel, msg.getUsername()); } else if (!this.allowAnonymous) { failedCredentials(channel); - return; + return false; } - - //if an old client with the same ID already exists close its session. - 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 - 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); - } - + return true; } - 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)); + private boolean sendAck(ConnectionDescriptor descriptor, ConnectMessage msg) { + final boolean success = descriptor.assignState(ConnectionState.DISCONNECTED, ConnectionState.SENDACK); + if (!success) { + return false; + } + doSendAck(descriptor.channel, msg); + return true; } - public void completeConnect(Channel channel, ConnectMessage msg) { - ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), channel, msg.isCleanSession()); - m_clientIDs.put(msg.getClientID(), connDescr); - - int keepAlive = msg.getKeepAlive(); - LOG.debug("Connect with keepAlive {} s", keepAlive); - NettyUtils.keepAlive(channel, keepAlive); - //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); - NettyUtils.cleanSession(channel, msg.isCleanSession()); - //used to track the client in the subscription and publishing phases. - //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); - NettyUtils.clientID(channel, msg.getClientID()); - LOG.debug("Connect create session <{}>", channel); - - setIdleTime(channel.pipeline(), Math.round(keepAlive * 1.5f)); - - //Handle will flag - if (msg.isWillFlag()) { - AbstractMessage.QOSType willQos = AbstractMessage.QOSType.valueOf(msg.getWillQos()); - byte[] willPayload = msg.getWillMessage(); - ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); - //save the will testament in the clientID store - WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos); - m_willStore.put(msg.getClientID(), will); - LOG.info("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic()); - } - + private void doSendAck(Channel channel, ConnectMessage msg) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); @@ -303,8 +288,46 @@ public class ProtocolProcessor { clientSession.cleanSession(msg.isCleanSession()); } channel.writeAndFlush(okResp); - m_interceptor.notifyClientConnected(msg); + } + private void initializeKeepAliveTimeout(Channel channel, ConnectMessage msg) { + int keepAlive = msg.getKeepAlive(); + LOG.debug("Connect with keepAlive {} s", keepAlive); + NettyUtils.keepAlive(channel, keepAlive); + //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); + NettyUtils.cleanSession(channel, msg.isCleanSession()); + //used to track the client in the subscription and publishing phases. + //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); + NettyUtils.clientID(channel, msg.getClientID()); + LOG.debug("Connect create session <{}>", channel); + + setIdleTime(channel.pipeline(), Math.round(keepAlive * 1.5f)); + } + + private void storeWillMessage(ConnectMessage msg) { + //Handle will flag + if (msg.isWillFlag()) { + AbstractMessage.QOSType willQos = AbstractMessage.QOSType.valueOf(msg.getWillQos()); + byte[] willPayload = msg.getWillMessage(); + ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); + //save the will testament in the clientID store + WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos); + m_willStore.put(msg.getClientID(), will); + LOG.info("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic()); + } + } + + private ClientSession createOrLoadClientSession(ConnectionDescriptor descriptor, ConnectMessage msg) { + final boolean success = descriptor.assignState(ConnectionState.SENDACK, ConnectionState.SESSION_CREATED); + if (!success) { + return null; + } + return doCreateSession(descriptor.channel, msg); + } + + private ClientSession doCreateSession(Channel channel, ConnectMessage msg) { + ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID()); + boolean isSessionAlreadyStored = clientSession != null; if (!isSessionAlreadyStored) { LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession()); @@ -314,16 +337,43 @@ public class ProtocolProcessor { if (msg.isCleanSession()) { clientSession.cleanSession(); } - LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); + LOG.info("Created session for client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); + return clientSession; + } + + private boolean republish(ConnectionDescriptor descriptor, ConnectMessage msg, ClientSession clientSession) { + final boolean success = descriptor.assignState(ConnectionState.SESSION_CREATED, ConnectionState.MESSAGES_REPUBLISHED); + if (!success) { + return false; + } + if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 republishStoredInSession(clientSession); } int flushIntervalMs = 500/*(keepAlive * 1000) / 2*/; - setupAutoFlusher(channel.pipeline(), flushIntervalMs); - LOG.info("CONNECT processed"); + setupAutoFlusher(descriptor.channel.pipeline(), flushIntervalMs); + return true; } + private void failedCredentials(Channel session) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); + session.writeAndFlush(okResp); + LOG.info("Client {} failed to connect with bad username or password.", session); + } + +// 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)); +// } + private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) { AutoFlushHandler autoFlushHandler = new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS); try { @@ -341,14 +391,6 @@ public class ProtocolProcessor { pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); } - private void failedCredentials(Channel session) { - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); - session.writeAndFlush(okResp); - session.close(); - LOG.info("Client {} failed to connect with bad username or password.", session); - } - /** * Republish QoS1 and QoS2 messages stored into the session for the clientID. * */ @@ -393,7 +435,7 @@ public class ProtocolProcessor { return; } String clientID = targetSession.clientID; - if (m_clientIDs.containsKey(clientID)) { + if (connectionDescriptors.containsKey(clientID)) { this.connectionsStatus.put(clientID, ConnectState.CONNECTED); } } @@ -602,17 +644,17 @@ public class ProtocolProcessor { } } - if (m_clientIDs == null) { - throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be " + + if (connectionDescriptors == null) { + throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " + "initialized, somewhere it's overwritten!!"); } - if (m_clientIDs.get(clientId) == null) { + if (connectionDescriptors.get(clientId) == null) { //TODO while we were publishing to the target client, that client disconnected, // could happen is not an error HANDLE IT throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", - clientId, m_clientIDs)); + clientId, connectionDescriptors)); } - Channel channel = m_clientIDs.get(clientId).channel; + Channel channel = connectionDescriptors.get(clientId).channel; LOG.trace("Session for clientId {}", clientId); if (channel.isWritable()) { LOG.debug("channel is writable"); @@ -629,7 +671,7 @@ public class ProtocolProcessor { LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); - m_clientIDs.get(clientID).channel.writeAndFlush(pubRecMessage); + connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage); } private void sendPubAck(String clientId, int messageID) { @@ -638,14 +680,14 @@ public class ProtocolProcessor { pubAckMessage.setMessageID(messageID); try { - if (m_clientIDs == null) { - throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); + if (connectionDescriptors == null) { + throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!"); } - LOG.debug("clientIDs are {}", m_clientIDs); - if (m_clientIDs.get(clientId) == null) { - throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); + LOG.debug("clientIDs are {}", connectionDescriptors); + if (connectionDescriptors.get(clientId) == null) { + throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors)); } - m_clientIDs.get(clientId).channel.writeAndFlush(pubAckMessage); + connectionDescriptors.get(clientId).channel.writeAndFlush(pubAckMessage); } catch(Throwable t) { LOG.error(null, t); } @@ -681,7 +723,7 @@ public class ProtocolProcessor { PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - m_clientIDs.get(clientID).channel.writeAndFlush(pubCompMessage); + connectionDescriptors.get(clientID).channel.writeAndFlush(pubCompMessage); } public void processPubRec(Channel channel, PubRecMessage msg) { @@ -738,7 +780,7 @@ public class ProtocolProcessor { //someone else changed the state (another connect with same client ID) return; } - m_clientIDs.remove(clientID); + connectionDescriptors.remove(clientID); //cleanup the will store m_willStore.remove(clientID); @@ -752,7 +794,7 @@ public class ProtocolProcessor { public void processConnectionLost(String clientID, boolean sessionStolen, Channel channel) { ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); - m_clientIDs.remove(clientID, oldConnDescr); + connectionDescriptors.remove(clientID, oldConnDescr); //If already removed a disconnect message was already processed for this clientID if (sessionStolen) { //de-activate the subscriptions for this 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 c79a122c..5433d644 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -481,8 +481,8 @@ public class ProtocolProcessorTest { }; processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR); //just to activate the two sessions - processor.m_clientIDs.put("Sub A", new ConnectionDescriptor("Sub A", null, true)); - processor.m_clientIDs.put("Sub B", new ConnectionDescriptor("Sub B", null, true)); + processor.connectionDescriptors.put("Sub A", new ConnectionDescriptor("Sub A", null, true)); + processor.connectionDescriptors.put("Sub B", new ConnectionDescriptor("Sub B", null, true)); //Exercise processor.route2Subscribers(forwardPublish);