diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java index 3447fa2a..16538949 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java @@ -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; 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 6f901cc3..92e3a231 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 @@ -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) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 85848203..5109cc89 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -38,6 +38,7 @@ public class HawtDBStorageService implements IStorageService { private SortedIndex m_inflightStore; //persistent Map of clientID, list of Subscriptions + //TODO move to Set to avoid double subscriptions private SortedIndex> m_persistentSubscriptions; public HawtDBStorageService() { @@ -161,8 +162,10 @@ public class HawtDBStorageService implements IStorageService { } List subs = m_persistentSubscriptions.get(clientID); - subs.add(newSubscription); - m_persistentSubscriptions.put(clientID, subs); + if (!subs.contains(newSubscription)) { + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); + } } public void removeAllSubscriptions(String clientID) { 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 7503c933..1958218b 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 @@ -20,11 +20,8 @@ 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.*; 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; @@ -63,7 +60,7 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, //private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); private RingBuffer m_ringBuffer; - private INotifier m_notifier; +// private INotifier m_notifier; private IStorageService m_storageService; @@ -71,27 +68,51 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, private IAuthenticator m_authenticator; - private ExecutorService m_executor = Executors.newFixedThreadPool(1); + private ExecutorService m_executor; BatchEventProcessor m_eventProcessor; + + private static SimpleMessaging INSTANCE; - public SimpleMessaging() { - m_ringBuffer = new RingBuffer(ValueEvent.EVENT_FACTORY, 1024 * 32); + private SimpleMessaging() { + /*m_ringBuffer = new RingBuffer(ValueEvent.EVENT_FACTORY, 1024 * 32); SequenceBarrier barrier = m_ringBuffer.newBarrier(); m_eventProcessor = new BatchEventProcessor(m_ringBuffer, barrier, this); m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence()); m_executor.submit(m_eventProcessor); + + disruptorPublish(new InitEvent()); */ - m_storageService = new HawtDBStorageService(); + /* m_storageService = new HawtDBStorageService(); m_storageService.initStore(); - subscriptions.init(m_storageService); + subscriptions.init(m_storageService); */ + } + + 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.EVENT_FACTORY, 1024 * 32); + + SequenceBarrier barrier = m_ringBuffer.newBarrier(); + m_eventProcessor = new BatchEventProcessor(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(); @@ -115,10 +136,10 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, 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); LOG.debug("subscribe invoked for topic: " + topic); - disruptorPublish(new SubscribeEvent(newSubscription)); + disruptorPublish(new SubscribeEvent(newSubscription, messageID)); } @@ -213,11 +234,16 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, processRepublish((RepublishEvent) evt); } else if (evt instanceof ConnectEvent) { processConnect((ConnectEvent) evt); + } else if (evt instanceof InitEvent) { + processInit(); } } - - public void stop() { - m_eventProcessor.halt(); + + private void processInit() { + m_storageService = new HawtDBStorageService(); + m_storageService.initStore(); + + subscriptions.init(m_storageService); } // protected void eventLoop() { @@ -383,16 +409,29 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, //scans retained messages to be published to the new subscription Collection 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) { @@ -423,6 +462,9 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, private void processClose() { LOG.debug("processClose invoked"); m_storageService.close(); + +// m_eventProcessor.halt(); + m_executor.shutdown(); } private void processRepublish(RepublishEvent evt) throws InterruptedException { @@ -432,7 +474,7 @@ public class SimpleMessaging implements IMessaging, EventHandler/*, } 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())); } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java new file mode 100644 index 00000000..94d7ce63 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java @@ -0,0 +1,6 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + */ +public class InitEvent extends MessagingEvent { +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java index 334e7e48..fba93dd8 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java @@ -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; + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index b86b90e5..d97efed9 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -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 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 8116064c..81ea59ec 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 @@ -7,6 +7,7 @@ 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 +15,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; @@ -68,7 +68,8 @@ 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); @@ -97,7 +98,7 @@ public class Server { LOG.info("Server stopping..."); messaging.close(); - messaging.stop(); + //messaging.stop(); // messagingEventLoop.interrupt(); // LOG.info("shutting down evet loop"); // try { @@ -113,7 +114,14 @@ public class Server { session.close(false); } } - + + //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()));*/ + + m_acceptor.unbind(); m_acceptor.dispose(); LOG.info("Server stopped"); 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 2c940bdb..b548791f 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 @@ -45,8 +45,9 @@ public class SimpleMessagingTest { 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); @@ -68,17 +69,21 @@ public class SimpleMessagingTest { } } }); + + //sleep to let the messaging batch processor to process the initEvent + Thread.sleep(300); } @After public void tearDown() { m_receivedMessage = null; + messaging.close(); } @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 @@ -88,7 +93,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 @@ -105,7 +110,7 @@ public class SimpleMessagingTest { 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)); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0); messaging.processSubscribe(evt); //Exercise @@ -116,13 +121,13 @@ public class SimpleMessagingTest { 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", "#")); @@ -130,8 +135,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", "+")); @@ -203,7 +208,7 @@ public class SimpleMessagingTest { //Exercise //m_handler.setMessaging(mockedMessaging); - SubscribeEvent pubEvt = new SubscribeEvent(subscription); + SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0); messaging.processSubscribe(pubEvt); //Verify