diff --git a/ChangeLog.txt b/ChangeLog.txt index 9350db12..48e6d0c8 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -6,7 +6,7 @@ Version 0.8: Fixed bug #68, UNSUBSCRIBE has to fail on bad topicFilter names Fixed bug #81, on stop of moquette broker it doesn't shutdown due to no shutdown of ringbuffer used on the back of ProtocolProcessor Moved config to be not only filesystem based, but also classpath and memory (fix issue #73) - Switched from commit pesistence on every change to time based to fix delay in handling many msgs/sec (fix issue #79) + Switched from commit persistence on every change to time based to fix delay in handling many msgs/sec (fix issue #79) 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 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 2de09f82..3a0e668a 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -36,7 +36,7 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.*; /** * MapDB main persistence implementation @@ -59,6 +59,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { private DB m_db; private String m_storePath; + private final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1); + /* * The default constructor will create an in memory store as no file path was specified */ @@ -92,6 +94,12 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); m_persistentSubscriptions = m_db.getHashMap("subscriptions"); m_qos2Store = m_db.getHashMap("qos2Store"); + m_scheduler.scheduleWithFixedDelay(new Runnable() { + @Override + public void run() { + m_db.commit(); + } + }, 30, 30, TimeUnit.SECONDS); } @Override @@ -110,7 +118,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { message.get(raw); m_retainedStore.put(topic, new StoredMessage(raw, qos, topic)); } - m_db.commit(); } @Override @@ -140,7 +147,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } storedEvents.add(convertToStored(evt)); m_persistentMessageStore.put(clientID, storedEvents); - m_db.commit(); //NB rewind the evt message content LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic()); } @@ -174,12 +180,10 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } events.remove(toRemoveEvt); m_persistentMessageStore.put(clientID, events); - m_db.commit(); } public void dropMessagesInSession(String clientID) { m_persistentMessageStore.remove(clientID); - m_db.commit(); } //----------------- In flight methods ----------------- @@ -191,7 +195,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { if (inFlightForClient != null) { inFlightForClient.remove(packetID); } - m_db.commit(); } @Override @@ -199,7 +202,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { String publishKey = String.format("%s%d", clientID, packetID); StoredPublishEvent storedEvt = convertToStored(evt); m_inflightStore.put(publishKey, storedEvt); - m_db.commit(); } /** @@ -241,7 +243,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { clientSubscriptions.remove(toBeRemoved); } m_persistentSubscriptions.put(clientID, clientSubscriptions); - m_db.commit(); } public void addNewSubscription(Subscription newSubscription) { @@ -270,18 +271,15 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_persistentSubscriptions.put(clientID, subs); LOG.debug("clientID {} subscriptions set now is {}", clientID, subs); } - m_db.commit(); } public void wipeSubscriptions(String clientID) { m_persistentSubscriptions.remove(clientID); - m_db.commit(); } @Override public void updateSubscriptions(String clientID, Set subscriptions) { m_persistentSubscriptions.put(clientID, subscriptions); - m_db.commit(); } public List listAllSubscriptions() { diff --git a/tools_scripts/integration/issue79_stress_multipub.groovy b/tools_scripts/integration/issue79_stress_multipub.groovy new file mode 100644 index 00000000..04adcf85 --- /dev/null +++ b/tools_scripts/integration/issue79_stress_multipub.groovy @@ -0,0 +1,55 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') +//@GrabResolver(name='Paho', root="https://repo.eclipse.org/content/repositories/paho-releases/") +//@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.2') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +class PublisherThread extends Thread { + + private final int pubId + private final String host = "localhost" + private final byte[] payload + private final int numToSend + + PublisherThread(int pubId, int numToSend, byte[] payload) { + this.numToSend = numToSend + this.payload = payload + this.pubId = pubId + } + + @Override + public void run() { + MQTT mqtt = new MQTT() + //mqtt.setHost("test.mosquitto.org", 1883); + mqtt.setHost(host, 1883) + mqtt.setCleanSession(true) + mqtt.setHost(host, 1883) + + mqtt.setClientId("PublisherClient${pubId}") + BlockingConnection connection = mqtt.blockingConnection() + connection.connect() + (1..numToSend).each { + connection.publish("/topic${pubId}", this.payload, QoS.AT_MOST_ONCE, false) + } + connection.publish("/exit${pubId}", 'Exit'.bytes, QoS.AT_MOST_ONCE, false) + connection.disconnect() + } +} + +if (args.size() < 1) { + println "Usage: groovy issue79_stress_multipub.groovy num_publishers" +} + +final byte[] payload = 'Hello world!!'.bytes + +int numPublishers = args[0] as Integer +(1..numPublishers).each { numPub -> + def pubTh = new PublisherThread(numPub, 5, payload) + pubTh.start() +} + +println "Started all publisher threads" \ No newline at end of file diff --git a/tools_scripts/integration/issue79_stress_multisub.groovy b/tools_scripts/integration/issue79_stress_multisub.groovy new file mode 100644 index 00000000..2a1602a0 --- /dev/null +++ b/tools_scripts/integration/issue79_stress_multisub.groovy @@ -0,0 +1,56 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') +//@GrabResolver(name='Paho', root="https://repo.eclipse.org/content/repositories/paho-releases/") +//@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.2') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +class SubscriberThread extends Thread { + + private final int subId + private final String host = "localhost" + + SubscriberThread(int subId) { + this.subId = subId + } + + @Override + public void run() { + MQTT mqtt = new MQTT() + //mqtt.setHost("test.mosquitto.org", 1883); + mqtt.setHost(host, 1883) + mqtt.setCleanSession(true) + mqtt.setHost(host, 1883) + + mqtt.setClientId("SubscriberClient${subId}") + BlockingConnection connection = mqtt.blockingConnection() + connection.connect() + Topic[] topics = [new Topic("/topic${subId}", QoS.AT_MOST_ONCE), new Topic("/exit${subId}", QoS.AT_MOST_ONCE)] + byte[] qoses = connection.subscribe(topics) + boolean exit = false + while (!exit) { + Message message = connection.receive() + if (message.topic == "/exit${subId}") { + exit = true + } + message.ack() + } + connection.disconnect() + println "sub ${subId} exited" + } +} + +if (args.size() < 1) { + println "Usage: groovy issue79_stress_multisub.groovy num_subscribers" +} + +int numSubscriber = args[0] as Integer +(1..numSubscriber).each { numSub -> + def subTh = new SubscriberThread(numSub) + subTh.start() +} + +println "Started all subscribers threads" \ No newline at end of file