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

Fixed bug due to not persisted subscriptions into the store, on restart there were ghosts subscriptions that produced errors

Browse Source
This commit is contained in:
selva.andre
2014-09-19 22:00:51 +02:00
parent cc45b87ebe
commit 0940ac84a0
11 changed files with 124 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+1
View File
@@ -2,6 +2,7 @@ Version 0.7:
Fix issue #38, added aligned to MQTT 3.1.1
Refactored big event switch to @Annotation dispatcher
Switched from HawtDB to MapDB
Fixed bug that didn't update client subscription on persistent store so that on server restart there where ghost subscriptions.
Version 0.6:
Fix issue #42, bad password file path composition
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java
+7
View File
@@ -16,6 +16,8 @@
package org.dna.mqtt.moquette.messaging.spi;
import java.util.List;
import java.util.Set;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
/**
@@ -35,6 +37,11 @@ public interface ISessionsStore {
*/
void wipeSubscriptions(String sessionID);
/**
* Updates the subscriptions set for the clientID
* */
void updateSubscriptions(String clientID, Set<Subscription> subscriptions);
List<Subscription> listAllSubscriptions();
/**
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+6 -1
View File
@@ -142,7 +142,12 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
public void wipeSubscriptions(String clientID) {
m_persistentSubscriptions.remove(clientID);
}
@Override
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
m_persistentSubscriptions.put(clientID, subscriptions);
}
@Override
public boolean contains(String clientID) {
return m_persistentSubscriptions.containsKey(clientID);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+10 -9
View File
@@ -19,7 +19,6 @@ import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -30,7 +29,6 @@ import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
import org.dna.mqtt.moquette.messaging.spi.persistence.MapDBPersistentStore;
import org.dna.mqtt.moquette.proto.messages.*;
@@ -61,6 +59,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private ISessionsStore m_sessionsStore;
private ExecutorService m_executor;
private Disruptor<ValueEvent> m_disruptor;
private static SimpleMessaging INSTANCE;
@@ -82,14 +81,14 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
public void init(Properties configProps) {
subscriptions = new SubscriptionsStore();
m_executor = Executors.newFixedThreadPool(1);
Disruptor<ValueEvent> disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
/*Disruptor<ValueEvent> disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor,
m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
/*Disruptor<ValueEvent> m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor,
ProducerType.MULTI, new BusySpinWaitStrategy());*/
disruptor.handleEventsWith(this);
disruptor.start();
m_disruptor.handleEventsWith(this);
m_disruptor.start();
// Get the ring buffer from the Disruptor to be used for publishing.
m_ringBuffer = disruptor.getRingBuffer();
m_ringBuffer = m_disruptor.getRingBuffer();
annotationSupport.processAnnotations(m_processor);
processInit(configProps);
@@ -127,6 +126,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
boolean elapsed = !m_stopLatch.await(10, TimeUnit.SECONDS);
LOG.debug("after m_stopLatch");
m_executor.shutdown();
m_disruptor.shutdown();
if (elapsed) {
LOG.error("Can't stop the server in 10 seconds");
}
@@ -164,8 +164,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
m_storageService.initStore();
List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
subscriptions.init(storedSubscriptions);
//List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
//subscriptions.init(storedSubscriptions);
subscriptions.init(m_sessionsStore);
String passwdPath = props.getProperty("password_file", "");
String configPath = System.getProperty("moquette.path", null);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java
+15 -9
View File
@@ -16,12 +16,10 @@
package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,7 +29,7 @@ import org.slf4j.LoggerFactory;
* @author andrea
*/
public class SubscriptionsStore {
public static interface IVisitor<T> {
void visit(TreeNode node);
@@ -70,19 +68,21 @@ public class SubscriptionsStore {
}
private TreeNode subscriptions = new TreeNode(null);
private ISessionsStore m_sessionsStore;
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
/**
* Initialize the subscription tree with the list of subscriptions.
*/
public void init(List<Subscription> subscriptions) {
public void init(ISessionsStore sessionsStore) {
LOG.debug("init invoked");
m_sessionsStore = sessionsStore;
List<Subscription> subscriptions = sessionsStore.listAllSubscriptions();
//reload any subscriptions persisted
if (LOG.isDebugEnabled()) {
LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree());
}
for (Subscription subscription : subscriptions) {
LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter());
addDirect(subscription);
@@ -169,18 +169,24 @@ public class SubscriptionsStore {
*/
public void removeForClient(String clientID) {
subscriptions.removeClientSubscriptions(clientID);
//TODO persist the update
//remove from log all subscriptions
// m_storageService.removeAllSubscriptions(clientID);
}
public void deactivate(String clientID) {
subscriptions.deactivate(clientID);
//persist the update
Set<Subscription> subs = subscriptions.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
public void activate(String clientID) {
LOG.debug("Activating subscriptions for clientID <{}>", clientID);
subscriptions.activate(clientID);
//persist the update
Set<Subscription> subs = subscriptions.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
/**
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java
+18 -6
View File
@@ -15,15 +15,11 @@
*/
package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Queue;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
class TreeNode {
private class ClientIDComparator implements Comparator<Subscription> {
public int compare(Subscription o1, Subscription o2) {
@@ -181,6 +177,22 @@ class TreeNode {
for (TreeNode child : m_children) {
child.activate(clientID);
}
}
/**
* @return the set of subscriptions for the given client.
* */
Set<Subscription> findAllByClientID(String clientID) {
Set<Subscription> subs = new HashSet<Subscription>();
for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
subs.add(s);
}
}
//go deep
for (TreeNode child : m_children) {
subs.addAll(child.findAllByClientID(clientID));
}
return subs;
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java
+7
View File
@@ -203,6 +203,12 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
m_db.commit();
}
@Override
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
m_persistentSubscriptions.put(clientID, subscriptions);
m_db.commit();
}
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
@@ -219,6 +225,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
public void close() {
this.m_db.commit();
LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
this.m_db.close();
LOG.debug("closed disk storage");
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+5 -5
View File
@@ -162,7 +162,7 @@ public class ProtocolProcessorTest {
m_mockAuthenticator = new MockAuthenticator(users);
subscriptions = new SubscriptionsStore();
subscriptions.init(Collections.<Subscription>emptyList());
subscriptions.init(new MemoryStorageService());
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator);
}
@@ -298,7 +298,7 @@ public class ProtocolProcessorTest {
};
//simulate a connect that register a clientID to an IoSession
subs.init(Collections.<Subscription>emptyList());
subs.init(new MemoryStorageService());
m_processor.init(subs, m_storageService, m_sessionStore, null);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProcotolVersion((byte)3);
@@ -342,7 +342,7 @@ public class ProtocolProcessorTest {
};
//simulate a connect that register a clientID to an IoSession
subs.init(Collections.<Subscription>emptyList());
subs.init(new MemoryStorageService());
m_processor.init(subs, m_storageService, m_sessionStore, null);
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
@@ -372,7 +372,7 @@ public class ProtocolProcessorTest {
m_processor.processPublish(m_session, msg);
//Verify
Thread.sleep(100); //ugly but we dependend on the asynch that pull data from back disruptor
Thread.sleep(100); //ugly but we depend on the asynch that pull data from back disruptor
PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverSession.getMessage();
assertNotNull(pub2FirstSubscriber);
String firstMessageContent = DebugUtils.payload2Str(pub2FirstSubscriber.getPayload());
@@ -458,7 +458,7 @@ public class ProtocolProcessorTest {
}
}
};
subs.init(Collections.<Subscription>emptyList());
subs.init(new MemoryStorageService());
//simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_storageService, m_sessionStore, null);
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java
+5 -4
View File
@@ -22,6 +22,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.dna.mqtt.moquette.messaging.spi.impl.MemoryStorageService;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.junit.After;
import static org.junit.Assert.*;
@@ -42,7 +43,7 @@ public class SubscriptionsStoreTest {
@Before
public void setUp() throws IOException {
store = new SubscriptionsStore();
store.init(Collections.<Subscription>emptyList());
store.init(new MemoryStorageService());
}
@Test
@@ -232,7 +233,7 @@ public class SubscriptionsStoreTest {
private void assertMatch(String subscription, String topic) {
store = new SubscriptionsStore();
store.init(Collections.<Subscription>emptyList());
store.init(new MemoryStorageService());
Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false);
store.add(sub);
assertFalse(store.matches(topic).isEmpty());
@@ -240,7 +241,7 @@ public class SubscriptionsStoreTest {
private void assertNotMatch(String subscription, String topic) {
store = new SubscriptionsStore();
store.init(Collections.<Subscription>emptyList());
store.init(new MemoryStorageService());
Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false);
store.add(sub);
assertTrue(store.matches(topic).isEmpty());
@@ -325,7 +326,7 @@ public class SubscriptionsStoreTest {
@Test
public void removeSubscription_withDifferentClients_subscribedSameTopic() {
SubscriptionsStore aStore = new SubscriptionsStore();
aStore.init(Collections.<Subscription>emptyList());
aStore.init(new MemoryStorageService());
//subscribe a not active clientID1 to /topic
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false);
aStore.add(slashSub);
broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java
+9 -4
View File
@@ -64,8 +64,10 @@ public class ServerLowlevelMessagesIntegrationTests {
@After
public void tearDown() throws Exception {
m_client.close();
LOG.debug("After raw client close");
Thread.sleep(300); //to let the close event pass before server stop event
m_server.stopServer();
LOG.debug("After asked server to stop");
File dbFile = new File(Server.STORAGE_FILE_PATH);
if (dbFile.exists()) {
dbFile.delete();
@@ -111,11 +113,14 @@ public class ServerLowlevelMessagesIntegrationTests {
//Execute
m_client.sendMessage(connectMessage);
long connectTime = System.currentTimeMillis();
//but after the 2 KEEP ALIVE timeut expires it gets fired,
//but after the 2 KEEP ALIVE timeout expires it gets fired,
//NB it's 1,5 * KEEP_ALIVE so 3 secs and some millis to propagate the message
Message msg = willSubscriber.receive(3100, TimeUnit.MILLISECONDS);
Message msg = willSubscriber.receive(3300, TimeUnit.MILLISECONDS);
long willMessageReceiveTime = System.currentTimeMillis();
if (msg == null) {
LOG.warn("testament message is null");
}
assertNotNull("the will message should be fired after keep alive!", msg);
msg.ack();
//the will message hasn't to be received before the elapsing of Keep Alive timeout
broker/src/test/java/org/dna/mqtt/moquette/server/ServerRestartIntegrationTest.java
+41 -4
View File
@@ -49,7 +49,9 @@ public class ServerRestartIntegrationTest {
startServer();
m_mqtt = new MQTT();
m_mqtt.setHost("localhost", 1883);
m_mqtt.setCleanSession(false);
}
@After
@@ -72,10 +74,8 @@ public class ServerRestartIntegrationTest {
@Test
public void checkRestartCleanSubscriptionTree() throws Exception {
//subscribe to /topic
m_mqtt.setHost("localhost", 1883);
m_mqtt.setCleanSession(false);
m_mqtt.setClientId("Subscriber");
//subscribe to /topic
m_subscriber = m_mqtt.blockingConnection();
m_subscriber.connect();
Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)};
@@ -98,7 +98,7 @@ public class ServerRestartIntegrationTest {
topics = new Topic[]{new Topic("/topic", QoS.EXACTLY_ONCE)};
m_subscriber.subscribe(topics);
//shoud be just one registration so a publisher receive one notification
//should be just one registration so a publisher receive one notification
MQTT mqtt = new MQTT();
mqtt.setHost("localhost", 1883);
mqtt.setClientId("Publisher");
@@ -113,4 +113,41 @@ public class ServerRestartIntegrationTest {
//no more messages on the same topic will be received
assertNull(m_subscriber.receive(1, TimeUnit.SECONDS));
}
@Test
public void checkDontPublishInactiveClientsAfterServerRestart() throws Exception {
m_mqtt.setClientId("SubPub");
BlockingConnection conn = subscribeAndPublish("/topic");
conn.disconnect();
//shutdown the server
m_server.stopServer();
//restart the server
m_server.startServer();
MQTT mqtt = new MQTT();
mqtt.setHost("localhost", 1883);
mqtt.setClientId("Publisher");
m_publisher = mqtt.blockingConnection();
m_publisher.connect();
m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false);
}
/**
* Connect subscribe to topic and publish on the same topic
* */
private BlockingConnection subscribeAndPublish(String topic) throws Exception {
BlockingConnection conn = m_mqtt.blockingConnection();
conn.connect();
Topic[] topics = new Topic[]{new Topic(topic, QoS.AT_MOST_ONCE)};
conn.subscribe(topics);
conn.publish(topic, "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false);
//read the message
Message msg = conn.receive();
msg.ack();
assertEquals("Hello world MQTT!!", new String(msg.getPayload()));
return conn;
}
}