Renamed close method and event to more clear Stop

This commit is contained in:
selva.andre
2012-09-03 21:59:00 +02:00
parent 870eb12390
commit 9d0aaa0eb5
5 changed files with 13 additions and 22 deletions
@@ -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,
@@ -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<ValueEvent>/*,
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<ValueEvent>/*,
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<ValueEvent>/*,
// } 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<ValueEvent>/*,
subscriptions.disconnect(clientID);
}
private void processClose() {
LOG.debug("processClose invoked");
private void processStop() {
LOG.debug("processStop invoked");
m_storageService.close();
// m_eventProcessor.halt();
@@ -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 {
}
@@ -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");
@@ -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