From 224bcc17219f4594bfff530a3795c4e9471ae645 Mon Sep 17 00:00:00 2001
From: andrea <andrea@samarcanda>
Date: Thu, 26 Dec 2013 21:10:34 +0100
Subject: [PATCH] Modified SimpleMessaging.close() to be synchronous and avoid
 thread synch problems in stop/start cycle used in integration tests. Fix
 issue #19

---
 .../messaging/spi/impl/SimpleMessaging.java       | 15 +++++++++++++++
 .../java/org/dna/mqtt/moquette/server/Server.java | 15 ++++++---------
 .../server/ServerIntegrationPahoTest.java         |  1 -
 3 files changed, 21 insertions(+), 10 deletions(-)

diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
index b0f8c2d6..c65a116c 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
@@ -4,8 +4,10 @@ import com.lmax.disruptor.BatchEventProcessor;
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.SequenceBarrier;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import org.dna.mqtt.moquette.messaging.spi.IMessaging;
 import org.dna.mqtt.moquette.messaging.spi.IStorageService;
 import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
@@ -42,6 +44,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     
     private ProtocolProcessor m_processor = new ProtocolProcessor();
     
+    CountDownLatch m_stopLatch;
+    
     private SimpleMessaging() {
     }
 
@@ -92,7 +96,17 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
 
     public void stop() {
+        m_stopLatch = new CountDownLatch(1);
         disruptorPublish(new StopEvent());
+        try {
+            //wait the callback notification from the protocol processor thread
+            boolean elapsed = !m_stopLatch.await(10, TimeUnit.SECONDS);
+            if (elapsed) {
+                LOG.error("Can't stop the server in 10 seconds");
+            }
+        } catch (InterruptedException ex) {
+            LOG.error(null, ex);
+        }
     }
     
     public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
@@ -182,5 +196,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         m_executor.shutdown();
         
         subscriptions = null;
+        m_stopLatch.countDown();
     }
 }
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
index cedab003..43332572 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
@@ -45,19 +45,16 @@ public class Server {
         m_acceptor.initialize(messaging);
     }
     
-    /**
-     * It's an asynchronous method, it's only a request to stop
-     */
     public void stopServer() {
         System.out.println("Server stopping...");
         messaging.stop();
         m_acceptor.close();
         System.out.println("Server stopped");
-        try {
-            //sleep one second to give the disruptor
-            Thread.sleep(100);
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+//        try {
+//            //sleep one second to give the disruptor
+//            Thread.sleep(100);
+//        } catch (InterruptedException ex) {
+//            LOG.error(null, ex);
+//        }
     }
 }
diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
index c810fd28..9cef50e9 100644
--- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
+++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
@@ -315,7 +315,6 @@ public class ServerIntegrationPahoTest {
         anotherClient.disconnect();
     }
 
-    //BUG
     @Test
     public void testPublishWithQoS2() throws Exception {
         LOG.info("*** testPublishWithQoS2 ***");