Upgraded from disruptor 2.7.3 to 3.3.0

broker/pom.xml

@@ -33,9 +33,9 @@
         </dependency>
         
         <dependency>
-            <groupId>com.googlecode.disruptor</groupId>
+            <groupId>com.lmax</groupId>
             <artifactId>disruptor</artifactId>
-            <version>2.7.1</version>
+            <version>3.3.0</version>
         </dependency>
         
         <dependency>

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -15,10 +15,9 @@
  */
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
-import com.lmax.disruptor.BatchEventProcessor;
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.SequenceBarrier;
+import com.lmax.disruptor.dsl.Disruptor;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.HashMap;
@@ -110,7 +109,6 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     private Map<String, WillMessage> m_willStore = new HashMap<String, WillMessage>();
     
     private ExecutorService m_executor;
-    BatchEventProcessor<ValueEvent> m_eventProcessor;
     private RingBuffer<ValueEvent> m_ringBuffer;
 
     ProtocolProcessor() {}
@@ -135,13 +133,12 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //init the output ringbuffer
         m_executor = Executors.newFixedThreadPool(1);
 
-        m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
+        Disruptor<ValueEvent> disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
+        disruptor.handleEventsWith(this);
+        disruptor.start();
 
-        SequenceBarrier barrier = m_ringBuffer.newBarrier();
+        // Get the ring buffer from the Disruptor to be used for publishing.
-        m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
+        m_ringBuffer = disruptor.getRingBuffer();
-        //TODO in a presentation is said to don't do the followinf line!!
-        m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
-        m_executor.submit(m_eventProcessor);
     }
     
     @MQTTMessage(message = ConnectMessage.class)
@@ -228,7 +225,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         }
         LOG.debug("processConnect sent OK ConnAck");
         session.write(okResp);
-        LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
+        LOG.warn("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
         
         LOG.info("Create persistent session for clientID {}", msg.getClientID());
         m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true), msg.getClientID()); //null means EmptySubscription
@@ -525,7 +522,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //cleanup the will store
         m_willStore.remove(clientID);
         
-        LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession);
+        LOG.warn("Disconnected client <{}> with clean session {}", clientID, cleanSession);
     }
     
     void proccessConnectionLost(String clientID) {

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -15,10 +15,11 @@
  */
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
-import com.lmax.disruptor.BatchEventProcessor;
+import com.lmax.disruptor.BusySpinWaitStrategy;
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.SequenceBarrier;
+import com.lmax.disruptor.SleepingWaitStrategy;
+import com.lmax.disruptor.dsl.Disruptor;
 
 import java.util.List;
 import java.util.Properties;
@@ -26,6 +27,8 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+
+import com.lmax.disruptor.dsl.ProducerType;
 import org.dna.mqtt.moquette.messaging.spi.IMessaging;
 import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
 import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
@@ -61,7 +64,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     private ISessionsStore m_sessionsStore;
 
     private ExecutorService m_executor;
-    BatchEventProcessor<ValueEvent> m_eventProcessor;
 
     private static SimpleMessaging INSTANCE;
     
@@ -83,15 +85,15 @@ 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,
+                ProducerType.MULTI, new BusySpinWaitStrategy());*/
+        disruptor.handleEventsWith(this);
+        disruptor.start();
 
-        m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
+        // Get the ring buffer from the Disruptor to be used for publishing.
+        m_ringBuffer = disruptor.getRingBuffer();
 
-        SequenceBarrier barrier = m_ringBuffer.newBarrier();
-        m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
-        //TODO in a presentation is said to don't do the following line!!
-        m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
-        m_executor.submit(m_eventProcessor);
-        
         annotationSupport.processAnnotations(m_processor);
         processInit(configProps);
 //        disruptorPublish(new InitEvent(configProps));

broker/src/test/resources/log4j.properties

@@ -5,7 +5,7 @@ log4j.rootLogger=ERROR, stdout, file
 log4j.logger.org.dna.mqtt=INFO
 log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG
 log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationQoSValidationTest=INFO
-log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO
+log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=WARN
 #log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=DEBUG
 #log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=TRACE
 log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging=WARN
@@ -25,7 +25,8 @@ log4j.appender.stdout.Threshold=DEBUG
 # for debug trace
 #log4j.appender.stdout.Threshold=TRACE 
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
+#log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n
 
 #file appender
 log4j.appender.file=org.apache.log4j.RollingFileAppender
@@ -34,4 +35,5 @@ log4j.appender.file.File=moquette.log
 log4j.appender.file.MaxFileSize=1MB
 log4j.appender.file.MaxBackupIndex=1
 log4j.appender.file.layout=org.apache.log4j.PatternLayout
-log4j.appender.file.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
+#log4j.appender.file.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
+log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n

bundle/pom.xml

@@ -37,12 +37,11 @@
             <artifactId>moquette-broker</artifactId>
             <version>${project.version}</version>
         </dependency>
-        
+
         <dependency>
-            <groupId>com.googlecode.disruptor</groupId>
+            <groupId>com.lmax</groupId>
             <artifactId>disruptor</artifactId>
-            <version>2.7.1</version>
+            <version>3.3.0</version>
-            <!--<version>3.2.0</version>-->
         </dependency>
         
         <dependency>

tools_scripts/publisher_client.groovy

@@ -17,7 +17,7 @@ MqttMessage message = new MqttMessage('Hello world!!'.bytes)
 message.setQos(0)
 print "publishing.."
 long startTime = System.currentTimeMillis()
-(1..10).each {
+(1..1000000).each {
     client.publish('topic', 'Hello world!!'.bytes, 0, false)
 }
 client.publish('/exit', 'Exit'.bytes, 0, false)