diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index fde4b2b8..a00e721b 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -24,8 +24,8 @@ import java.util.concurrent.atomic.AtomicReference; /** * Value object to maintain the information of single connection, like ClientID, Channel, * and clean session flag. - * - * + * + * * @author andrea */ public class ConnectionDescriptor { @@ -38,12 +38,12 @@ public class ConnectionDescriptor { //Disconnection states SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED; } - + 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; this.channel = session; @@ -51,9 +51,13 @@ public class ConnectionDescriptor { } 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(); + LOG.info("closing the channel"); +// try { + //this.channel.disconnect().sync(); + this.channel.close();//.sync(); +// } catch (InterruptedException e) { +// e.printStackTrace(); +// } } public boolean assignState(ConnectionState expected, ConnectionState newState) { 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 2017bb5b..4a2e216c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -96,9 +96,48 @@ public class ProtocolProcessor { } + private enum SubscriptionState { + STORED, VERIFIED + //Connection states + //DISCONNECTED, SENDACK, SESSION_CREATED, MESSAGES_REPUBLISHED, ESTABLISHED, + //Disconnection states + //SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED; + } + + private class RunningSubscription { + final String clientID; + final long packeId; + + RunningSubscription(String clientID, long packeId) { + this.clientID = clientID; + this.packeId = packeId; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + RunningSubscription that = (RunningSubscription) o; + + if (packeId != that.packeId) return false; + return clientID != null ? clientID.equals(that.clientID) : that.clientID == null; + + } + + @Override + public int hashCode() { + int result = clientID != null ? clientID.hashCode() : 0; + result = 31 * result + (int) (packeId ^ (packeId >>> 32)); + return result; + } + } + private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); protected ConcurrentMap connectionDescriptors; + protected ConcurrentMap subscriptionInCourse; + private SubscriptionsStore subscriptions; private boolean allowAnonymous; private boolean allowZeroByteClientId; @@ -129,13 +168,6 @@ public class ProtocolProcessor { init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, allowZeroByteClientId, authorizator,interceptor,null); } - void init(SubscriptionsStore subscriptions, IMessagesStore storageService, - ISessionsStore sessionsStore, - IAuthenticator authenticator, - boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { - init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, interceptor, serverPort); - } - /** * @param subscriptions the subscription store where are stored all the existing * clients subscriptions. @@ -154,6 +186,7 @@ public class ProtocolProcessor { boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { this.connectionDescriptors = new ConcurrentHashMap<>(); + this.subscriptionInCourse = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; this.allowAnonymous = allowAnonymous; @@ -265,11 +298,6 @@ public class ProtocolProcessor { if (!success) { return false; } - doSendAck(descriptor.channel, msg); - return true; - } - - private void doSendAck(Channel channel, ConnectMessage msg) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); @@ -281,7 +309,8 @@ public class ProtocolProcessor { if (isSessionAlreadyStored) { clientSession.cleanSession(msg.isCleanSession()); } - channel.writeAndFlush(okResp); + descriptor.channel.writeAndFlush(okResp); + return true; } private void initializeKeepAliveTimeout(Channel channel, ConnectMessage msg) { @@ -316,10 +345,7 @@ public class ProtocolProcessor { 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) { @@ -867,40 +893,35 @@ public class ProtocolProcessor { public void processSubscribe(Channel channel, SubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); LOG.info("SUBSCRIBE client <{}>", clientID); - LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, msg.getMessageID()); - - ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - //ack the client - SubAckMessage ackMessage = new SubAckMessage(); - ackMessage.setMessageID(msg.getMessageID()); + int messageID = msg.getMessageID(); + LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, messageID); + RunningSubscription executionKey = new RunningSubscription(clientID, messageID); + SubscriptionState currentStatus = this.subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); + if (currentStatus != null) { + LOG.debug("The client <{}> sent another SUBSCRIBE while this one was processing", clientID); + return; + } String username = NettyUtils.userName(channel); - List newSubscriptions = new ArrayList<>(); - for (SubscribeMessage.Couple req : msg.subscriptions()) { - if (!m_authorizator.canRead(req.topicFilter, username, clientSession.clientID)) { - //send SUBACK with 0x80, the user hasn't credentials to read the topic - LOG.debug("topic {} doesn't have read credentials", req.topicFilter); - ackMessage.addType( AbstractMessage.QOSType.FAILURE); - continue; - } - - AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.qos); - Subscription newSubscription = new Subscription(clientID, req.topicFilter, qos); - boolean valid = clientSession.subscribe(newSubscription); - ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE); - if (valid) { - newSubscriptions.add(newSubscription); - } + List ackTopics = doVerify(clientID, username, msg); + SubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics); + if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { + LOG.debug("The client {} sent another SUBSCRIBE while this one was verifing topicFilters"); + return; } + ackMessage.setMessageID(messageID); + List newSubscriptions = doStoreSubscription(ackTopics, clientID); + //save session, persist subscriptions from session - LOG.debug("SUBACK for packetID {}", msg.getMessageID()); + LOG.debug("SUBACK for packetID {}", messageID); if (LOG.isTraceEnabled()) { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } for (Subscription subscription : newSubscriptions) { - subscribeSingleTopic(subscription); + LOG.debug("Persisting subscription {}", subscription); + subscriptions.add(subscription.asClientTopicCouple()); } channel.writeAndFlush(ackMessage); @@ -908,13 +929,69 @@ public class ProtocolProcessor { for (Subscription subscription : newSubscriptions) { publishStoredMessagesInSession(subscription, username); } + + boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); + if (!success) { + LOG.warn("Failed to remove the descriptor, something bad happened"); + } } - private void subscribeSingleTopic(final Subscription newSubscription) { - LOG.debug("Subscribing {}", newSubscription); - subscriptions.add(newSubscription.asClientTopicCouple()); + private List doStoreSubscription(List ackTopics, String clientID) { + ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); + + List newSubscriptions = new ArrayList<>(); + for (SubscribeMessage.Couple req : ackTopics) { + //TODO this is SUPER UGLY + if (req.qos == AbstractMessage.QOSType.FAILURE.byteValue()) { + continue; + } + AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.qos); + Subscription newSubscription = new Subscription(clientID, req.topicFilter, qos); + clientSession.subscribe(newSubscription); + newSubscriptions.add(newSubscription); + } + return newSubscriptions; } + /** + * @return the list of verified topics + * @param clientID + * @param username + * */ + private List doVerify(String clientID, String username, SubscribeMessage msg) { + ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); + List ackTopics = new ArrayList<>(); + + for (SubscribeMessage.Couple req : msg.subscriptions()) { + if (!m_authorizator.canRead(req.topicFilter, username, clientSession.clientID)) { + //send SUBACK with 0x80, the user hasn't credentials to read the topic + LOG.debug("topic {} doesn't have read credentials", req.topicFilter); + ackTopics.add(new SubscribeMessage.Couple(AbstractMessage.QOSType.FAILURE.byteValue(), req.topicFilter)); + } else { + boolean validTopic = SubscriptionsStore.validate(req.topicFilter); + AbstractMessage.QOSType qos = validTopic ? + AbstractMessage.QOSType.valueOf(req.qos) + : AbstractMessage.QOSType.FAILURE; + ackTopics.add(new SubscribeMessage.Couple(qos.byteValue(), req.topicFilter)); + } + } + return ackTopics; + } + + /** + * Create the SUBACK response from a list of topicFilters + * */ + private SubAckMessage doAckMessageFromValidateFilters(List topicFilters) { + //ack the client + SubAckMessage ackMessage = new SubAckMessage(); + for (SubscribeMessage.Couple req : topicFilters) { + ackMessage.addType(AbstractMessage.QOSType.valueOf(req.qos)); + } + return ackMessage; + } + + + private void publishStoredMessagesInSession(final Subscription newSubscription, String username) { LOG.debug("Publish persisted messages in session {}", newSubscription); diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java index a7550020..879b62a0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java @@ -85,6 +85,7 @@ class TreeNode { m_subscriptions.remove(clientTopicCouple); } + //TODO smell a query method that return the result modifing the parameter (matchingSubs) void matches(Queue tokens, List matchingSubs) { Token t = tokens.poll(); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index ef780315..f0149293 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -474,13 +474,19 @@ public class ServerIntegrationPahoTest { LOG.info("*** testForceClientDisconnection_issue118 ***"); MessageCollector cbSubscriber1 = new MessageCollector(); MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1); + LOG.info("Connected 'subscriber' first time"); clientXA.subscribe("topic", 0); + LOG.info("Subscribed 'topic' from 'subscriber' first time"); MqttClient clientXB = createClient("publisher", "X"); + LOG.info("Connected 'publisher' first time"); clientXB.publish("topic", "Hello".getBytes(), 2, false); + LOG.info("Published on 'topic' from 'publisher' first time"); + LOG.info("Creating second new 'subscriber'"); MessageCollector cbSubscriber2 = new MessageCollector(); MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2); + LOG.info("Connected 'subscriber' second time"); clientYA.subscribe("topic", 0); MqttClient clientYB = createClient("publisher", "Y"); 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 5433d644..6e674ab8 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -215,6 +215,7 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); + msg.setMessageID(10); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_channel, msg); @@ -236,6 +237,7 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); + msg.setMessageID(10); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_channel, msg); @@ -251,6 +253,7 @@ public class ProtocolProcessorTest { @Test public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); + msg.setMessageID(10); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -270,6 +273,7 @@ public class ProtocolProcessorTest { @Test public void testSubscribeWithBadFormattedTopic() { SubscribeMessage msg = new SubscribeMessage(); + msg.setMessageID(10); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -344,6 +348,7 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); + msg.setMessageID(10); msg.addSubscription(new SubscribeMessage.Couple(QOSType.MOST_ONE.byteValue(), "#")); m_processor.processSubscribe(m_channel, msg); 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 147e0526..bf55c538 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 @@ -317,6 +317,7 @@ public class ProtocolProcessor_CONNECT_Test { //subscribe SubscribeMessage subscribeMsg = new SubscribeMessage(); + subscribeMsg.setMessageID(10); subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), ProtocolProcessorTest.FAKE_TOPIC)); NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID);