anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Changed the way a Client session is active and moved to event and states of Channel for CONNECT and DISCONNET messages

Browse Source
This commit is contained in:
andsel
2016-11-05 18:29:08 +01:00
parent 8e43f75f70
commit 7508275cf7
6 changed files with 107 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+11
View File
@@ -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);
}
}
}
broker/src/main/java/io/moquette/server/netty/NettyUtils.java
+11
View File
@@ -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<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
private static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
private static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
private static final AttributeKey<Object> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
private static final AttributeKey<Object> ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN);
private static final AttributeKey<Object> ATTR_KEY_CHANNEL_STATUS = AttributeKey.valueOf(ATTR_CHANNEL_STATUS);
public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey<Object> key) {
Attribute<Object> 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();
}
}
broker/src/main/java/io/moquette/spi/ClientSession.java
+2 -16
View File
@@ -61,8 +61,6 @@ public class ClientSession {
private volatile boolean cleanSession;
private boolean active = false;
private BlockingQueue<AbstractMessage> 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);
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+79 -17
View File
@@ -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<String, ConnectionDescriptor> 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<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
ProtocolProcessor() {}
private final ConcurrentMap<String, ConnectState> 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
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+3 -3
View File
@@ -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);
broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java
+1 -1
View File
@@ -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());