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 7e5fcf92..1450f4a0 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java @@ -82,14 +82,4 @@ public class NettyUtils { public static Boolean sessionStolen(Channel channel) { return (Boolean) channel.attr(NettyUtils.ATTR_KEY_SESSION_STOLEN).get(); } - - @Deprecated - public static void channelStatus(Channel channel, ProtocolProcessor.ConnectState newStatus) { - channel.attr(NettyUtils.ATTR_KEY_CHANNEL_STATUS).set(newStatus); - } - - @Deprecated - 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/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index ce706d5d..663df965 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -98,10 +98,6 @@ public class ProtocolProcessor { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - public enum ConnectState { - CONNECTED, DISCONNECTING, DISCONNECTED, DROP_BY_CONNECT - } - protected ConcurrentMap connectionDescriptors; private SubscriptionsStore subscriptions; private boolean allowAnonymous; @@ -116,9 +112,6 @@ 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() {} public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, @@ -333,8 +326,6 @@ public class ProtocolProcessor { LOG.debug("Create persistent session for clientID <{}>", msg.getClientID()); clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession()); } - this.connectionsStatus.put(msg.getClientID(), ConnectState.CONNECTED); - NettyUtils.channelStatus(channel, ConnectState.CONNECTED); if (msg.isCleanSession()) { clientSession.cleanSession(); } @@ -364,17 +355,6 @@ public class ProtocolProcessor { 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 { @@ -422,7 +402,6 @@ public class ProtocolProcessor { //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(targetSession); StoredMessage inflightMsg = targetSession.getInflightMessage(messageID); targetSession.inFlightAcknowledged(messageID); @@ -431,16 +410,6 @@ public class ProtocolProcessor { m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); } - private void verifyToActivate(ClientSession targetSession) { - if (targetSession == null) { - return; - } - String clientID = targetSession.clientID; - if (connectionDescriptors.containsKey(clientID)) { - this.connectionsStatus.put(clientID, ConnectState.CONNECTED); - } - } - private static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) { IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName()); stored.setRetained(msg.isRetainFlag()); @@ -592,9 +561,7 @@ public class ProtocolProcessor { qos = sub.getRequestedQos(); } ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); - verifyToActivate(targetSession); -// boolean targetIsActive = this.connectionsStatus.get(sub.getClientId()) == ConnectState.CONNECTED; boolean targetIsActive = this.connectionDescriptors.containsKey(sub.getClientId()); LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", @@ -704,7 +671,6 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(targetSession); IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); route2Subscribers(evt); @@ -731,7 +697,6 @@ public class ProtocolProcessor { public void processPubRec(Channel channel, PubRecMessage msg) { String clientID = NettyUtils.clientID(channel); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(targetSession); //remove from the inflight and move to the QoS2 second phase queue int messageID = msg.getMessageID(); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); @@ -750,7 +715,6 @@ public class ProtocolProcessor { LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(targetSession); StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); String username = NettyUtils.userName(channel); String topic = inflightMsg.getTopic(); @@ -798,36 +762,6 @@ public class ProtocolProcessor { return; } -// ConnectState status = NettyUtils.channelStatus(channel); -// if (status != null && status == ConnectState.DROP_BY_CONNECT) { -// LOG.info("DISCONNECT drop by other connect"); -// return; -// } -// 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(); -// 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; -// } -// connectionDescriptors.remove(clientID); -// -// //cleanup the will store -// m_willStore.remove(clientID); -// -// String username = NettyUtils.userName(channel); -// m_interceptor.notifyClientDisconnected(clientID, username); -// this.connectionsStatus.remove(clientID); LOG.info("DISCONNECT client <{}> finished", clientID); } @@ -908,13 +842,11 @@ public class ProtocolProcessor { */ public void processUnsubscribe(Channel channel, UnsubscribeMessage msg) { List topics = msg.topicFilters(); - int messageID = msg.getMessageID(); String clientID = NettyUtils.clientID(channel); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientSession); for (String topic : topics) { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { @@ -931,6 +863,7 @@ public class ProtocolProcessor { } //ack the client + int messageID = msg.getMessageID(); UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(messageID); @@ -944,7 +877,6 @@ public class ProtocolProcessor { LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientSession); //ack the client SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); @@ -1004,7 +936,6 @@ public class ProtocolProcessor { LOG.debug("Found {} messages to republish", messages.size()); ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); - verifyToActivate(targetSession); for (IMessagesStore.StoredMessage storedMsg : messages) { //fire as retained the message LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter());