diff --git a/ChangeLog.txt b/ChangeLog.txt index cef57fa2..0e6b1d66 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -10,6 +10,8 @@ Version 0.8: Changed willMessage and password in CONNECT message from String to byte[] (fix issue #94 and #95) Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription Added check to avoid creation of LostConnectionEvents for not yet completed connects (fix issue #88) + Removed LMAX ringbuffer to use multi threading with persistent (copy on write) data structures + Fixed bug that on connection stolen dropped also the new connection (fix issue #108) Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 0e1caeb9..43174b56 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -274,7 +274,6 @@ public class ProtocolProcessor { private void cleanSession(String clientID) { LOG.info("cleaning old saved subscriptions for client <{}>", clientID); //remove from log all subscriptions - //m_sessionsStore.wipeSubscriptions(clientID); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -578,7 +577,6 @@ public class ProtocolProcessor { } subscriptions.removeSubscription(topic, clientID); - m_sessionsStore.removeSubscription(topic, clientID); m_interceptor.notifyTopicUnsubscribed(topic, clientID); } @@ -619,7 +617,6 @@ public class ProtocolProcessor { //send SUBACK with 0x80 for this topic filter return false; } - m_sessionsStore.addNewSubscription(newSubscription); subscriptions.add(newSubscription); //notify the Observables diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index fb78242c..f8cef25b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -179,6 +179,7 @@ public class SubscriptionsStore { } public void add(Subscription newSubscription) { + m_sessionsStore.addNewSubscription(newSubscription); addDirect(newSubscription); } @@ -205,6 +206,7 @@ public class SubscriptionsStore { } //spin lock repeating till we can, swap root, if can't swap just re-do the operation } while(!subscriptions.compareAndSet(oldRoot, couple.root)); + m_sessionsStore.removeSubscription(topic, clientID); } /** @@ -213,7 +215,7 @@ public class SubscriptionsStore { public void clearAllSubscriptions() { SubscriptionTreeCollector subsCollector = new SubscriptionTreeCollector(); bfsVisit(subscriptions.get(), subsCollector, 0); - + List allSubscriptions = subsCollector.getResult(); for (Subscription subscription : allSubscriptions) { removeSubscription(subscription.getTopicFilter(), subscription.getClientId()); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java new file mode 100644 index 00000000..0648bbaa --- /dev/null +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java @@ -0,0 +1,101 @@ +/* + * Copyright (c) 2012-2015 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.spi.impl; + +import org.eclipse.moquette.spi.ISessionsStore; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; + +/** + * + * @author andrea + */ +public class MemorySessionStore implements ISessionsStore { + private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); + + private Map> m_persistentSubscriptions = new HashMap<>(); + + @Override + public void removeSubscription(String topic, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); + if (!m_persistentSubscriptions.containsKey(clientID)) { + return; + } + Set clientSubscriptions = m_persistentSubscriptions.get(clientID); + //search for the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : clientSubscriptions) { + if (sub.getTopicFilter().equals(topic)) { + toBeRemoved = sub; + break; + } + } + + if (toBeRemoved != null) { + clientSubscriptions.remove(toBeRemoved); + } + } + + @Override + public void addNewSubscription(Subscription newSubscription) { + final String clientID = newSubscription.getClientId(); + if (!m_persistentSubscriptions.containsKey(clientID)) { + m_persistentSubscriptions.put(clientID, new HashSet()); + } + + Set subs = m_persistentSubscriptions.get(clientID); + if (!subs.contains(newSubscription)) { + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); + } + } + + @Override + public void wipeSubscriptions(String clientID) { + m_persistentSubscriptions.remove(clientID); + } + + @Override + public void updateSubscriptions(String clientID, Set subscriptions) { + m_persistentSubscriptions.put(clientID, subscriptions); + } + + @Override + public boolean contains(String clientID) { + return m_persistentSubscriptions.containsKey(clientID); + } + + @Override + public void createNewSession(String clientID) { + if (m_persistentSubscriptions.containsKey(clientID)) { + LOG.error("already exists a session for client <{}>", clientID); + return; + } + m_persistentSubscriptions.put(clientID, new HashSet()); + } + + @Override + public List listAllSubscriptions() { + List allSubscriptions = new ArrayList(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + allSubscriptions.addAll(entry.getValue()); + } + return allSubscriptions; + } +} diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 0efe2c8e..a249045e 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -459,6 +459,7 @@ public class ProtocolProcessorTest { Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false); Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false); SubscriptionsStore subscriptions = new SubscriptionsStore(); + subscriptions.init(new MemorySessionStore()); subscriptions.add(subQos1); subscriptions.add(subQos2);