diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 92e3a231..ce1958af 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -37,7 +37,7 @@ public interface IMessaging { */ void removeSubscriptions(String clientID); - void close(); + void stop(); /** * Remove the clientID from topic subscription, if not previously subscribed, diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 1958218b..0648b923 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -16,7 +16,6 @@ import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; @@ -208,8 +207,8 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, disruptorPublish(new RemoveAllSubscriptionsEvent(clientID)); } - public void close() { - disruptorPublish(new CloseEvent()); + public void stop() { + disruptorPublish(new StopEvent()); } public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { @@ -223,8 +222,8 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, processUnsubscribe((UnsubscribeEvent) evt); } else if (evt instanceof RemoveAllSubscriptionsEvent) { processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt); - } else if (evt instanceof CloseEvent) { - processClose(); + } else if (evt instanceof StopEvent) { + processStop(); } else if (evt instanceof DisconnectEvent) { processDisconnect((DisconnectEvent) evt); } else if (evt instanceof CleanInFlightEvent) { @@ -261,7 +260,7 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, // } else if (evt instanceof RemoveAllSubscriptionsEvent) { // processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt); // } else if (evt instanceof CloseEvent) { -// processClose(); +// processStop(); // } else if (evt instanceof DisconnectEvent) { // processDisconnect((DisconnectEvent) evt); // } else if (evt instanceof CleanInFlightEvent) { @@ -459,8 +458,8 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, subscriptions.disconnect(clientID); } - private void processClose() { - LOG.debug("processClose invoked"); + private void processStop() { + LOG.debug("processStop invoked"); m_storageService.close(); // m_eventProcessor.halt(); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java similarity index 51% rename from parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java rename to parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java index 8ca2119e..d422abc5 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java @@ -1,9 +1,10 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; /** + * Poison pill to shutdown the event loop * * @author andrea */ -public class CloseEvent extends MessagingEvent { +public class StopEvent extends MessagingEvent { } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index 81ea59ec..ea2563dc 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -3,9 +3,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; + import org.apache.mina.core.service.IoAcceptor; import org.apache.mina.core.service.IoServiceStatistics; import org.apache.mina.core.session.IdleStatus; @@ -97,7 +95,7 @@ public class Server { protected void stopServer() { LOG.info("Server stopping..."); - messaging.close(); + messaging.stop(); //messaging.stop(); // messagingEventLoop.interrupt(); // LOG.info("shutting down evet loop"); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index b548791f..c7d834f5 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -1,13 +1,9 @@ package org.dna.mqtt.moquette.messaging.spi.impl; -import java.util.concurrent.BlockingQueue; - import org.apache.mina.core.filterchain.IoFilterAdapter; import org.apache.mina.core.session.DummySession; import org.apache.mina.core.session.IoSession; import org.apache.mina.core.write.WriteRequest; -import org.dna.mqtt.moquette.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; @@ -18,14 +14,11 @@ import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; -import org.dna.mqtt.moquette.server.MQTTHandler; import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.mockito.ArgumentCaptor; import static org.junit.Assert.assertEquals; -import static org.mockito.Mockito.*; /** * @@ -77,7 +70,7 @@ public class SimpleMessagingTest { @After public void tearDown() { m_receivedMessage = null; - messaging.close(); + messaging.stop(); } @Test