anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'change_from_blocking_queue_to_disruptor'

Browse Source
This commit is contained in:
selva.andre
2012-09-04 10:15:52 +02:00
parent c69802ef91 ab65cd250c
commit 7d5e91afae
15 changed files with 422 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
+1 -7
View File
@@ -4,13 +4,7 @@ import java.io.*;
import java.net.URISyntaxException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.fusesource.mqtt.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
parser/pom.xml
+6
View File
@@ -25,6 +25,12 @@
<version>2.0.4</version>
</dependency>
<dependency>
<groupId>com.googlecode.disruptor</groupId>
<artifactId>disruptor</artifactId>
<version>2.7.1</version>
</dependency>
<!-- <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
parser/src/main/java/org/dna/mqtt/moquette/client/Client.java
+1 -1
View File
@@ -63,7 +63,7 @@ public final class Client {
private int m_connectRetries = DEFAULT_RETRIES;
private String m_hostname;
private int m_port;
//internal mangement used for conversation with the server
//internal management used for conversation with the server
private IoConnector m_connector;
private IoSession m_session;
private CountDownLatch m_connectBarrier;
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+2 -2
View File
@@ -29,7 +29,7 @@ public interface IMessaging {
/**
* Subscribe a client to a specified topic with a defined level
*/
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession);
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID);
/**
* Remove all subscription to any topic the client (identified by clientID)
@@ -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,
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+12 -13
View File
@@ -13,10 +13,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.*;
import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage;
@@ -38,7 +35,8 @@ public class HawtDBStorageService implements IStorageService {
private SortedIndex<String, PublishEvent> m_inflightStore;
//persistent Map of clientID, list of Subscriptions
private SortedIndex<String, List<Subscription>> m_persistentSubscriptions;
//TODO move to Set to avoid double subscriptions
private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
public HawtDBStorageService() {
String storeFile = Server.STORAGE_FILE_PATH;
@@ -84,11 +82,10 @@ public class HawtDBStorageService implements IStorageService {
}
private void initPersistentSubscriptions() {
BTreeIndexFactory<String, List<Subscription>> indexFactory =
new BTreeIndexFactory<String, List<Subscription>>();
BTreeIndexFactory<String, Set<Subscription>> indexFactory = new BTreeIndexFactory<String, Set<Subscription>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentSubscriptions = (SortedIndex<String, List<Subscription>>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory);
m_persistentSubscriptions = (SortedIndex<String, Set<Subscription>>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory);
}
/**
@@ -157,12 +154,14 @@ public class HawtDBStorageService implements IStorageService {
public void addNewSubscription(Subscription newSubscription, String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
m_persistentSubscriptions.put(clientID, new ArrayList<Subscription>());
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
List<Subscription> subs = m_persistentSubscriptions.get(clientID);
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
if (!subs.contains(newSubscription)) {
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
}
}
public void removeAllSubscriptions(String clientID) {
@@ -171,7 +170,7 @@ public class HawtDBStorageService implements IStorageService {
public List<Subscription> retrieveAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
for (Map.Entry<String, List<Subscription>> entry : m_persistentSubscriptions) {
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions) {
allSubscriptions.addAll(entry.getValue());
}
return allSubscriptions;
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+125 -118
View File
@@ -1,38 +1,37 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
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.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.IAuthenticator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import java.io.Serializable;
import java.text.ParseException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.dna.mqtt.moquette.proto.messages.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.IAuthenticator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging, Runnable {
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*, Runnable*/ {
//TODO probably move this
public static class StoredMessage implements Serializable {
@@ -57,9 +56,9 @@ public class SimpleMessaging implements IMessaging, Runnable {
private SubscriptionsStore subscriptions = new SubscriptionsStore();
private BlockingQueue<MessagingEvent> m_inboundQueue = new LinkedBlockingQueue<MessagingEvent>();
private RingBuffer<ValueEvent> m_ringBuffer;
private INotifier m_notifier;
// private INotifier m_notifier;
private IStorageService m_storageService;
@@ -67,87 +66,86 @@ public class SimpleMessaging implements IMessaging, Runnable {
private IAuthenticator m_authenticator;
public SimpleMessaging() {
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
private ExecutorService m_executor;
BatchEventProcessor<ValueEvent> m_eventProcessor;
subscriptions.init(m_storageService);
private static SimpleMessaging INSTANCE;
private SimpleMessaging() {
}
public static SimpleMessaging getInstance() {
if (INSTANCE == null) {
INSTANCE = new SimpleMessaging();
}
return INSTANCE;
}
public void init() {
m_executor = Executors.newFixedThreadPool(1);
m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
SequenceBarrier barrier = m_ringBuffer.newBarrier();
m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
m_executor.submit(m_eventProcessor);
disruptorPublish(new InitEvent());
}
public void setNotifier(INotifier notifier) {
/* public void setNotifier(INotifier notifier) {
m_notifier= notifier;
}
} */
public void run() {
eventLoop();
private void disruptorPublish(MessagingEvent msgEvent) {
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
event.setEvent(msgEvent);
m_ringBuffer.publish(sequence);
}
public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) {
try {
m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, session));
}
public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session) {
try {
m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID, session));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, messageID, session));
}
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession) {
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID) {
Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession);
try {
LOG.debug("subscribe invoked for topic: " + topic);
m_inboundQueue.put(new SubscribeEvent(newSubscription));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
LOG.debug("subscribe invoked for topic: " + topic);
disruptorPublish(new SubscribeEvent(newSubscription, messageID));
}
public void unsubscribe(String topic, String clientID) {
try {
m_inboundQueue.put(new UnsubscribeEvent(topic, clientID));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
disruptorPublish(new UnsubscribeEvent(topic, clientID));
}
public void disconnect(IoSession session) {
try {
m_inboundQueue.put(new DisconnectEvent(session));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
disruptorPublish(new DisconnectEvent(session));
}
//method used by hte Notifier to re-put an event on the inbound queue
private void refill(MessagingEvent evt) throws InterruptedException {
m_inboundQueue.put(evt);
disruptorPublish(evt);
}
public void republishStored(String clientID) {
//create the event to push
try {
m_inboundQueue.put(new RepublishEvent(clientID));
} catch (InterruptedException iex) {
LOG.error(null, iex);
}
disruptorPublish(new RepublishEvent(clientID));
}
public void connect(IoSession session, ConnectMessage msg) {
try {
m_inboundQueue.put(new ConnectEvent(session, msg));
} catch (InterruptedException iex) {
LOG.error(null, iex);
}
disruptorPublish(new ConnectEvent(session, msg));
}
/**
@@ -189,56 +187,49 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
public void removeSubscriptions(String clientID) {
try {
m_inboundQueue.put(new RemoveAllSubscriptionsEvent(clientID));
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
disruptorPublish(new RemoveAllSubscriptionsEvent(clientID));
}
public void close() {
try {
m_inboundQueue.put(new CloseEvent());
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
public void stop() {
disruptorPublish(new StopEvent());
}
protected void eventLoop() {
LOG.debug("Started event loop");
boolean interrupted = false;
while (!interrupted) {
try {
MessagingEvent evt = m_inboundQueue.take();
if (evt instanceof PublishEvent) {
processPublish((PublishEvent) evt);
} else if (evt instanceof SubscribeEvent) {
processSubscribe((SubscribeEvent) evt);
} else if (evt instanceof UnsubscribeEvent) {
processUnsubscribe((UnsubscribeEvent) evt);
} else if (evt instanceof RemoveAllSubscriptionsEvent) {
processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt);
} else if (evt instanceof CloseEvent) {
processClose();
} else if (evt instanceof DisconnectEvent) {
processDisconnect((DisconnectEvent) evt);
} else if (evt instanceof CleanInFlightEvent) {
//remove the message from inflight storage
m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
} else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
} else if (evt instanceof ConnectEvent) {
processConnect((ConnectEvent) evt);
}
} catch (InterruptedException ex) {
processClose();
interrupted = true;
}
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
LOG.debug("onEvent processing event");
MessagingEvent evt = t.getEvent();
if (evt instanceof PublishEvent) {
processPublish((PublishEvent) evt);
} else if (evt instanceof SubscribeEvent) {
processSubscribe((SubscribeEvent) evt);
} else if (evt instanceof UnsubscribeEvent) {
processUnsubscribe((UnsubscribeEvent) evt);
} else if (evt instanceof RemoveAllSubscriptionsEvent) {
processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt);
} else if (evt instanceof StopEvent) {
processStop();
} else if (evt instanceof DisconnectEvent) {
processDisconnect((DisconnectEvent) evt);
} else if (evt instanceof CleanInFlightEvent) {
//remove the message from inflight storage
m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
} else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
} else if (evt instanceof ConnectEvent) {
processConnect((ConnectEvent) evt);
} else if (evt instanceof InitEvent) {
processInit();
}
}
private void processConnect(ConnectEvent evt) {
private void processInit() {
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
subscriptions.init(m_storageService);
}
protected void processConnect(ConnectEvent evt) {
ConnectMessage msg = evt.getMessage();
IoSession session = evt.getSession();
@@ -368,16 +359,29 @@ public class SimpleMessaging implements IMessaging, Runnable {
//scans retained messages to be published to the new subscription
Collection<StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return matchTopics(key, topic); //To change body of implemented methods use File | Settings | File Templates.
return matchTopics(key, topic);
}
});
for (StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true));
}
// //ack the client
// SubAckMessage ackMessage = new SubAckMessage();
// ackMessage.setMessageID(evt.getMessageID());
//
// //TODO by now it handles only QoS 0 messages
// /*for (int i = 0; i < evt.subscriptions().size(); i++) {
// ackMessage.addType(QOSType.MOST_ONE);
// } */
// LOG.info("replying with SubAct to MSG ID {0}", evt.getMessageID());
// //session.write(ackMessage);
// IoSession session = m_clientIDs.get(evt.getSubscription().clientId).getSession();
// session.write(ackMessage).awaitUninterruptibly();
}
protected void processUnsubscribe(UnsubscribeEvent evt) {
@@ -405,9 +409,12 @@ public class SimpleMessaging implements IMessaging, Runnable {
subscriptions.disconnect(clientID);
}
private void processClose() {
LOG.debug("processClose invoked");
private void processStop() {
LOG.debug("processStop invoked");
m_storageService.close();
// m_eventProcessor.halt();
m_executor.shutdown();
}
private void processRepublish(RepublishEvent evt) throws InterruptedException {
@@ -417,7 +424,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
for (PublishEvent pubEvt : publishedEvents) {
m_notifier.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID()));
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ValueEvent.java
+29
View File
@@ -0,0 +1,29 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import com.lmax.disruptor.EventFactory;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
/**
* Carrier value object for the RingBuffer.
*
* @author andrea
*/
public final class ValueEvent {
private MessagingEvent m_event;
public MessagingEvent getEvent() {
return m_event;
}
public void setEvent(MessagingEvent event) {
m_event = event;
}
public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>() {
public ValueEvent newInstance() {
return new ValueEvent();
}
};
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java
+6
View File
@@ -0,0 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
*/
public class InitEvent extends MessagingEvent {
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java → parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java
+2 -1
View File
@@ -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 {
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
+8 -2
View File
@@ -9,13 +9,19 @@ import org.dna.mqtt.moquette.messaging.spi.impl.Subscription;
public class SubscribeEvent extends MessagingEvent {
Subscription m_subscription;
int m_messageID;
public SubscribeEvent(Subscription subscription) {
public SubscribeEvent(Subscription subscription, int messageID) {
m_subscription = subscription;
m_messageID = messageID;
}
public Subscription getSubscription() {
return m_subscription;
}
public int getMessageID() {
return m_messageID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+1 -1
View File
@@ -75,7 +75,7 @@ public class MQTTHandler extends IoHandlerAdapter/* implements INotifier*/ {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()],
cleanSession);
cleanSession, msg.getMessageID());
}
//ack the client
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+26 -18
View File
@@ -3,10 +3,9 @@ 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;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
@@ -14,7 +13,6 @@ import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.proto.*;
import org.dna.mqtt.moquette.proto.messages.*;
import org.slf4j.Logger;
@@ -36,7 +34,7 @@ public class Server {
//private static final int NOTIFIER_POOL_SIZE = 1;
private IoAcceptor m_acceptor;
SimpleMessaging messaging;
Thread messagingEventLoop;
// Thread messagingEventLoop;
//private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
public static void main(String[] args) throws IOException {
@@ -68,19 +66,22 @@ public class Server {
m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
MQTTHandler handler = new MQTTHandler();
messaging = new SimpleMessaging();
messaging = SimpleMessaging.getInstance();
messaging.init();
//TODO fix this hugly wiring
handler.setMessaging(messaging);
// messaging.setNotifier(handler);
messagingEventLoop = new Thread(messaging);
messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
messagingEventLoop.start();
// messagingEventLoop = new Thread(messaging);
// messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
// messagingEventLoop.start();
m_acceptor.setHandler(handler);
((NioSocketAcceptor)m_acceptor).setReuseAddress(true);
((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true);
m_acceptor.getSessionConfig().setReadBufferSize( 2048 );
m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
m_acceptor.getStatistics().setThroughputCalculationInterval(10);
m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis());
m_acceptor.bind( new InetSocketAddress(PORT) );
LOG.info("Server binded");
@@ -96,23 +97,30 @@ public class Server {
protected void stopServer() {
LOG.info("Server stopping...");
messaging.close();
messagingEventLoop.interrupt();
LOG.info("shutting down evet loop");
try {
messagingEventLoop.join();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
messaging.stop();
//messaging.stop();
// messagingEventLoop.interrupt();
// LOG.info("shutting down evet loop");
// try {
// messagingEventLoop.join();
// } catch (InterruptedException ex) {
// LOG.error(null, ex);
// }
/*m_notifierPool.shutdown();*/
//log statistics
IoServiceStatistics statistics = m_acceptor.getStatistics();
statistics.updateThroughput(System.currentTimeMillis());
System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput()));
System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput()));
for(IoSession session: m_acceptor.getManagedSessions().values()) {
if(session.isConnected() && !session.isClosing()){
session.close(false);
}
}
m_acceptor.unbind();
m_acceptor.dispose();
LOG.info("Server stopped");
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+138 -22
View File
@@ -1,18 +1,24 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.concurrent.BlockingQueue;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent;
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.impl.events.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import static org.junit.Assert.*;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
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.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
/**
*
@@ -24,15 +30,53 @@ public class SimpleMessagingTest {
final static String FAKE_TOPIC = "/news";
SimpleMessaging messaging;
byte m_returnCode;
IoSession m_session;
// MQTTHandler m_handler;
ConnectMessage connMsg;
AbstractMessage m_receivedMessage;
@Before
public void setUp() {
messaging = new SimpleMessaging();
public void setUp() throws InterruptedException {
messaging = SimpleMessaging.getInstance();
messaging.init();
connMsg = new ConnectMessage();
connMsg.setProcotolVersion((byte) 0x03);
m_session = new DummySession();
m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(NextFilter nextFilter, IoSession session,
WriteRequest writeRequest) throws Exception {
try {
m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
if (m_receivedMessage instanceof ConnAckMessage) {
ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
m_returnCode = buf.getReturnCode();
}
} catch (Exception ex) {
throw new AssertionError("Wrong return code");
}
}
});
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
}
@After
public void tearDown() {
m_receivedMessage = null;
messaging.stop();
}
@Test
public void testSubscribe() {
//Exercise
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
//Verify
@@ -42,7 +86,7 @@ public class SimpleMessagingTest {
@Test
public void testDoubleSubscribe() {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
//Exercise
@@ -55,26 +99,28 @@ public class SimpleMessagingTest {
@Test
public void testPublish() throws InterruptedException {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
//simulate a connect that register a clientID to an IoSession
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
INotifier notifier = mock(INotifier.class);
messaging.setNotifier(notifier);
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
messaging.processPublish(pubEvt);
//Verify
ArgumentCaptor<NotifyEvent> argument = ArgumentCaptor.forClass(NotifyEvent.class);
verify(notifier).notify(argument.capture());
assertNotNull(m_receivedMessage);
//TODO check received message attributes
}
@Test
public void testMatchTopics_simple() {
assertTrue(messaging.matchTopics("/", "/"));
assertTrue(messaging.matchTopics("/finance", "/finance"));
}
@Test
public void testMatchTopics_multi() {
assertTrue(messaging.matchTopics("finance", "#"));
@@ -82,8 +128,8 @@ public class SimpleMessagingTest {
assertTrue(messaging.matchTopics("finance/stock", "finance/#"));
assertTrue(messaging.matchTopics("finance/stock/ibm", "finance/#"));
}
@Test
public void testMatchTopics_single() {
assertTrue(messaging.matchTopics("finance", "+"));
@@ -94,4 +140,74 @@ public class SimpleMessagingTest {
assertTrue(messaging.matchTopics("/finance/stock/ibm", "/finance/+/ibm"));
assertFalse(messaging.matchTopics("/finance/stock", "+"));
}
@Test
public void testHandleConnect_BadProtocol() {
connMsg.setProcotolVersion((byte) 0x02);
//Exercise
ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
messaging.processConnect(pubEvt);
//Verify
assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
}
@Test
public void testConnect_badClientID() {
connMsg.setClientID("extremely_long_clientID_greater_than_23");
//Exercise
ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
messaging.processConnect(pubEvt);
//Verify
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
}
@Test
public void testWill() {
connMsg.setClientID("123");
connMsg.setWillFlag(true);
connMsg.setWillTopic("topic");
connMsg.setWillMessage("Topic message");
//IMessaging mockedMessaging = mock(IMessaging.class);
//Exercise
//m_handler.setMessaging(mockedMessaging);
ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
messaging.processConnect(pubEvt);
//Verify
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
//TODO verify the call
/*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
}
@Test
public void testHandleSubscribe() {
String topicName = "/news";
SubscribeMessage subscribeMsg = new SubscribeMessage();
subscribeMsg.setMessageID(0x555);
subscribeMsg.addSubscription(new SubscribeMessage.Couple(
(byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName));
//IMessaging mockedMessaging = mock(IMessaging.class);
m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
m_session.setAttribute(Constants.CLEAN_SESSION, false);
Subscription subscription = new Subscription("fakeID", "/news", QOSType.EXACTLY_ONCE, true);
//Exercise
//m_handler.setMessaging(mockedMessaging);
SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0);
messaging.processSubscribe(pubEvt);
//Verify
/*ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false));
assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());*/
//TODO verify the message ACK and corresponding QoS!!
}
}
parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java
-120
View File
@@ -1,120 +0,0 @@
package org.dna.mqtt.moquette.server;
import org.mockito.ArgumentCaptor;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.apache.mina.core.write.WriteRequest;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.session.DummySession;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
/**
*
* @author andrea
*/
public class MQTTHandlerTest {
byte m_returnCode;
IoSession m_session;
MQTTHandler m_handler;
ConnectMessage connMsg;
@Before
public void setUp() {
m_handler = new MQTTHandler();
connMsg = new ConnectMessage();
connMsg.setProcotolVersion((byte) 0x03);
m_session = new DummySession();
m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(NextFilter nextFilter, IoSession session,
WriteRequest writeRequest) throws Exception {
try {
ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage();
m_returnCode = buf.getReturnCode();
} catch (Exception ex) {
throw new AssertionError("Wrong return code");
}
}
});
}
@Test
public void testHandleConnect_BadProtocol() {
connMsg.setProcotolVersion((byte) 0x02);
//Exercise
m_handler.handleConnect(m_session, connMsg);
//Verify
assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
}
@Test
public void testConnect_badClientID() {
connMsg.setClientID("extremely_long_clientID_greater_than_23");
//Exercise
m_handler.handleConnect(m_session, connMsg);
//Verify
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
}
@Test
public void testWill() {
connMsg.setClientID("123");
connMsg.setWillFlag(true);
connMsg.setWillTopic("topic");
connMsg.setWillMessage("Topic message");
IMessaging mockedMessaging = mock(IMessaging.class);
//Exercise
m_handler.setMessaging(mockedMessaging);
m_handler.handleConnect(m_session, connMsg);
//Verify
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));
}
@Test
public void testHandleSubscribe() {
String topicName = "/news";
SubscribeMessage subscribeMsg = new SubscribeMessage();
subscribeMsg.setMessageID(0x555);
subscribeMsg.addSubscription(new SubscribeMessage.Couple(
(byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName));
IMessaging mockedMessaging = mock(IMessaging.class);
m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
m_session.setAttribute(Constants.CLEAN_SESSION, false);
//Exercise
m_handler.setMessaging(mockedMessaging);
m_handler.handleSubscribe(m_session, subscribeMsg);
//Verify
ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false));
assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());
//TODO verify the message ACK and corresponding QoS!!
}
// @Test
// public void testHandleSubscribeMultipleTopics() {
// fail("TOBE implemented");
// }
}
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+65 -59
View File
@@ -48,29 +48,29 @@ public class ServerIntegrationTest {
Client client = new Client("localhost", Server.PORT);
// Client client = new Client("test.mosquitto.org", 1883);
client.connect();
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
received = true;
}
});
client.publish("/topic", "Test my payload".getBytes());
client.close();
client.shutdown();
assertTrue(received);
}
@Test
public void testCleanSession_maintainClientSubscriptions() {
Client client = new Client("localhost", Server.PORT/*, "CLID_123"*/);
client.connect(false); //without session cleanup
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
@@ -78,24 +78,24 @@ public class ServerIntegrationTest {
}
});
client.close();
//Exercise that the client maintain the subscriptions
client.connect(false);
client.connect(false);
client.publish("/topic", "Test my payload".getBytes());
client.close();
//Verify
assertTrue(received);
//TearDown
//TearDown
client.shutdown();
}
@Test
public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() {
Client client = new Client("localhost", Server.PORT, "CLID_123");
client.connect(false); //without session cleanup
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
@@ -104,7 +104,7 @@ public class ServerIntegrationTest {
});
client.close();
client.shutdown();
//Exercise that the client maintain the subscriptions
client = new Client("localhost", Server.PORT, "CLID_123");
client.register("/topic", new IPublishCallback() {
@@ -113,20 +113,20 @@ public class ServerIntegrationTest {
received = true;
}
});
client.connect(false);
client.connect(false);
client.publish("/topic", "Test my payload".getBytes());
client.close();
//Verify
assertTrue(received);
//TearDown
//TearDown
client.shutdown();
}
}
/**
* Check that after a client has connected with clean session false, subscribed
* Check that after a client has connected with clean session false, subscribed
* to some topic and exited, if it reconnect with clean session true, the server
* correctly cleanup every previous subscription
*/
@@ -134,7 +134,7 @@ public class ServerIntegrationTest {
public void testCleanSession_correctlyClientSubscriptions() {
Client client = new Client("localhost", Server.PORT, "CLID_123");
client.connect(false); //without session cleanup
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
@@ -143,7 +143,7 @@ public class ServerIntegrationTest {
});
client.close();
client.shutdown();
//Exercise that the client maintain the subscriptions
client = new Client("localhost", Server.PORT, "CLID_123");
client.register("/topic", new IPublishCallback() {
@@ -152,16 +152,16 @@ public class ServerIntegrationTest {
received = true;
}
});
client.connect(true);
client.connect(true);
client.publish("/topic", "Test my payload".getBytes());
client.close();
//Verify
assertFalse(received);
//TearDown
//TearDown
client.shutdown();
}
}
@Test
@@ -174,21 +174,27 @@ public class ServerIntegrationTest {
public void published(String topic, byte[] message) {
received = true;
barrier.countDown();
}
});
//never called because no notification is pending
assertFalse(barrier.await(1, TimeUnit.SECONDS));
client.close();
client.shutdown();
server.stopServer();
server.startServer();
final CountDownLatch barrier2 = new CountDownLatch(1);
//Exercise that the client maintain the subscriptions
client = new Client("localhost", Server.PORT, "CLID_123");
client.register("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
received = true;
barrier.countDown();
barrier2.countDown();
}
});
client.connect(false);
@@ -196,7 +202,7 @@ public class ServerIntegrationTest {
client.close();
//Verify
barrier.await(1, TimeUnit.SECONDS);
assertTrue(barrier2.await(1, TimeUnit.SECONDS));
assertTrue(received);
//TearDown
@@ -209,14 +215,14 @@ public class ServerIntegrationTest {
final CountDownLatch barrier = new CountDownLatch(1);
Client client = new Client("localhost", Server.PORT, "CLID_123");
client.connect();
client.publish("/topic", "Test my payload".getBytes(), true);
client.close();
client.shutdown();
//Exercise that the client maintain the subscriptions
client = new Client("localhost", Server.PORT, "CLID_123");
client.connect();
client.connect();
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
@@ -225,24 +231,24 @@ public class ServerIntegrationTest {
}
});
client.close();
//Verify
barrier.await(1, TimeUnit.SECONDS);
assertTrue(received);
//TearDown
//TearDown
client.shutdown();
}
}
@Ignore
public void testUnsubscribe_do_not_notify_anymore_same_session() throws InterruptedException {
Client client = new Client("localhost", Server.PORT, "CLID_123");
// Client client = new Client("test.mosquitto.org", 1883);
client.connect();
final CountDownLatch barrier = new CountDownLatch(1);
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
@@ -250,33 +256,33 @@ public class ServerIntegrationTest {
barrier.countDown();
}
});
client.publish("/topic", "Test my payload".getBytes());
//wait 1 second to receive the published message
boolean unlocked = barrier.await(1000, TimeUnit.MILLISECONDS);
assertTrue(unlocked); //we were unlocked by the message reception
assertTrue(received);
//reinit the flag
received = false;
//unsubscrbe and republish to check no notification is raised up
client.unsubscribe("/topic");
client.publish("/topic", "Test my payload".getBytes());
assertFalse(received);
client.close();
client.shutdown();
}
@Test
public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException {
Client client = new Client("localhost", Server.PORT, "CLID_123");
// Client client = new Client("test.mosquitto.org", 1883);
client.connect();
final CountDownLatch barrier = new CountDownLatch(1);
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
@@ -284,33 +290,33 @@ public class ServerIntegrationTest {
barrier.countDown();
}
});
client.publish("/topic", "Test my payload".getBytes());
//wait 1 second to receive the published message
boolean unlocked = barrier.await(1000, TimeUnit.MILLISECONDS);
assertTrue(unlocked); //we were unlocked by the message reception
assertTrue(received);
//unsubscrbe and republish to check no notification is raised up
client.unsubscribe("/topic");
client.close();
client.shutdown();
//Exercise that the client maintain the subscriptions
client = new Client("localhost", Server.PORT, "CLID_123");
client.connect();
client.connect();
//reinit the flag
received = false;
final CountDownLatch barrier2 = new CountDownLatch(1);
client.publish("/topic", "Test my payload".getBytes());
unlocked = barrier2.await(1000, TimeUnit.MILLISECONDS);
assertFalse(unlocked); //we were unlocked by the timeout exipration, no message received
assertFalse(received);
client.close();
client.shutdown();
}