From d2940741f83d67dfd26fbce605aea699a645f7f8 Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:35:42 +0200 Subject: [PATCH 01/10] Fixed a TU to wait an asych publish event --- .../org/dna/mqtt/moquette/server/ServerIntegrationTest.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index e26e8373..49020886 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -163,7 +163,8 @@ public class ServerIntegrationTest { @Test - public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException { + public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException, InterruptedException { + final CountDownLatch barrier = new CountDownLatch(1); Client client = new Client("localhost", Server.PORT, "CLID_123"); client.connect(false); //without session cleanup @@ -185,6 +186,7 @@ public class ServerIntegrationTest { public void published(String topic, byte[] message) { received = true; + barrier.countDown(); } }); client.connect(false); @@ -192,6 +194,7 @@ public class ServerIntegrationTest { client.close(); //Verify + barrier.await(1, TimeUnit.SECONDS); assertTrue(received); //TearDown From f0064357f5d149eca2e9b57dd47567a7f30717e8 Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:36:26 +0200 Subject: [PATCH 02/10] Minor change added a log --- .../mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index 59ffe6bc..2c8da612 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -200,6 +200,7 @@ public class SubscriptionsStore { * client's topics subscriptions */ public void init(MultiIndexFactory multiFactory) { + LOG.debug("init invoked"); BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); indexFactory.setKeyCodec(StringCodec.INSTANCE); @@ -210,6 +211,7 @@ public class SubscriptionsStore { LOG.debug("Reloading all stored subscriptions..."); for (Map.Entry> entry : m_persistent) { for (Subscription subscription : entry.getValue()) { + LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic()); addDirect(subscription); } } From 56431a161ee5a60c6812583eaee955b186b376d5 Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:36:51 +0200 Subject: [PATCH 03/10] Minor change removed old commented code --- .../spi/impl/SubscriptionsStore.java | 24 ------------------- 1 file changed, 24 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index 2c8da612..72e5c25b 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -219,30 +219,6 @@ public class SubscriptionsStore { } protected void addDirect(Subscription newSubscription) { -// List tokens = new ArrayList(); -// try { -// tokens = splitTopic(newSubscription.topic); -// } catch (ParseException ex) { -// //TODO handle the parse exception -// LOG.error(null, ex); -//// return; -// } -// -// TreeNode current = subscriptions; -// for (Token token : tokens) { -// TreeNode matchingChildren; -// -// //check if a children with the same token already exists -// if ((matchingChildren = current.childWithToken(token)) != null) { -// current = matchingChildren; -// } else { -// //create a new node for the newly inserted token -// matchingChildren = new TreeNode(current); -// matchingChildren.setToken(token); -// current.addChild(matchingChildren); -// current = matchingChildren; -// } -// } TreeNode current = findMatchingNode(newSubscription.topic); current.addSubcription(newSubscription); } From 739f63bc40ca3427f2222f1728aaeec87438d3cb Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:48:21 +0200 Subject: [PATCH 04/10] Changed SimpleMessaging to be event driven (Incoming requests) --- .../messaging/spi/impl/SimpleMessaging.java | 166 +++++++++++++----- .../messaging/spi/impl/events/CloseEvent.java | 9 + .../spi/impl/events/MessagingEvent.java | 10 ++ .../spi/impl/events/PublishEvent.java | 37 ++++ .../events/RemoveAllSubscriptionsEvent.java | 19 ++ .../spi/impl/events/SubscribeEvent.java | 21 +++ .../spi/impl/events/UnsubscribeEvent.java | 25 +++ .../org/dna/mqtt/moquette/server/Server.java | 12 +- 8 files changed, 250 insertions(+), 49 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java 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 d59a5074..f58b759a 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 @@ -1,17 +1,23 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import java.io.File; import java.io.IOException; import java.io.Serializable; import java.text.ParseException; import java.util.List; import java.util.Map; -import java.util.concurrent.locks.ReadWriteLock; -import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; +import org.dna.mqtt.moquette.messaging.spi.impl.events.CloseEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.UnsubscribeEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; @@ -27,8 +33,8 @@ import org.slf4j.LoggerFactory; * * @author andrea */ -public class SimpleMessaging implements IMessaging { - +public class SimpleMessaging implements IMessaging, Runnable { + private static class StoredMessage implements Serializable { QOSType m_qos; byte[] m_payload; @@ -49,11 +55,10 @@ public class SimpleMessaging implements IMessaging { private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); - //TODO is poorly performace on concurrent load, use at least ReadWriteLock - - private ReadWriteLock rwLock = new ReentrantReadWriteLock(); private SubscriptionsStore subscriptions = new SubscriptionsStore(); + private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); + private INotifier m_notifier; private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; @@ -89,6 +94,10 @@ public class SimpleMessaging implements IMessaging { m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); } + public void run() { + eventLoop(); + } + public void setNotifier(INotifier notifier) { m_notifier = notifier; } @@ -97,54 +106,28 @@ public class SimpleMessaging implements IMessaging { //TODO for this moment no qos > 0 trivial implementation QOSType defQos = QOSType.MOST_ONE; - rwLock.readLock().lock(); try { - for (Subscription sub : subscriptions.matches(topic)) { - m_notifier.notify(sub.clientId, topic, defQos, message, false); - } - if (retain) { - if (message.length == 0) { - //clean the message from topic - m_retainedStore.remove(topic); - } else { - //store the message to the topic - m_retainedStore.put(topic, new StoredMessage(message, qos)); - } - } - } finally { - rwLock.readLock().unlock(); + m_inboundQueue.put(new PublishEvent(topic, defQos, message, retain)); + } catch (InterruptedException ex) { + LOG.error(null, ex); } } public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); - rwLock.writeLock().lock(); try { - subscriptions.add(newSubscription); - - //scans retained messages to be published to the new subscription - LOG.debug("Scanning all retained messages..."); - for (Map.Entry entry : m_retainedStore) { - StoredMessage storedMsg = entry.getValue(); - if (matchTopics(entry.getKey(), topic)) { - //fire the as retained the message - m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), - storedMsg.getPayload(), true); - } - } - } finally { - LOG.debug("Finished firing retained messages"); - rwLock.writeLock().unlock(); + m_inboundQueue.put(new SubscribeEvent(newSubscription)); + } catch (InterruptedException ex) { + LOG.error(null, ex); } } public void unsubscribe(String topic, String clientID) { - rwLock.writeLock().lock(); try { - subscriptions.removeSubscription(topic, clientID); - } finally { - rwLock.writeLock().unlock(); + m_inboundQueue.put(new UnsubscribeEvent(topic, clientID)); + } catch (InterruptedException ex) { + LOG.error(null, ex); } } @@ -180,24 +163,111 @@ public class SimpleMessaging implements IMessaging { } } + /** + * NOT SAFE Method, to be removed because used only in tests + */ protected SubscriptionsStore getSubscriptions() { return subscriptions; } public void removeSubscriptions(String clientID) { - rwLock.writeLock().lock(); - subscriptions.removeForClient(clientID); - rwLock.writeLock().unlock(); + try { + m_inboundQueue.put(new RemoveAllSubscriptionsEvent(clientID)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } } public void close() { - rwLock.writeLock().lock(); + try { + m_inboundQueue.put(new CloseEvent()); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } + + + 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(); + } + } catch (InterruptedException ex) { + processClose(); + interrupted = true; + } + } + } + + protected void processPublish(PublishEvent evt) { + LOG.debug("processPublish invoked"); + String topic = evt.getTopic(); + QOSType qos = evt.getQos(); + byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + for (Subscription sub : subscriptions.matches(topic)) { + m_notifier.notify(sub.clientId, topic, qos, message, false); + } + if (retain) { + if (message.length == 0) { + //clean the message from topic + m_retainedStore.remove(topic); + } else { + //store the message to the topic + m_retainedStore.put(topic, new StoredMessage(message, qos)); + } + } + } + + + protected void processSubscribe(SubscribeEvent evt) { + LOG.debug("processSubscribe invoked"); + Subscription newSubscription = evt.getSubscription(); + String topic = newSubscription.getTopic(); + + subscriptions.add(newSubscription); + + //scans retained messages to be published to the new subscription + LOG.debug("Scanning all retained messages..."); + for (Map.Entry entry : m_retainedStore) { + StoredMessage storedMsg = entry.getValue(); + if (matchTopics(entry.getKey(), topic)) { + //fire the as retained the message + m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), + storedMsg.getPayload(), true); + } + } + } + + protected void processUnsubscribe(UnsubscribeEvent evt) { + LOG.debug("processSubscribe invoked"); + subscriptions.removeSubscription(evt.getTopic(), evt.getClientID()); + } + + protected void processRemoveAllSubscriptions(RemoveAllSubscriptionsEvent evt) { + LOG.debug("processRemoveAllSubscriptions invoked"); + subscriptions.removeForClient(evt.getClientID()); + } + + private void processClose() { + LOG.debug("processClose invoked"); try { pageFactory.close(); } catch (IOException ex) { LOG.error(null, ex); - } finally { - rwLock.writeLock().unlock(); } } } 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/CloseEvent.java new file mode 100644 index 00000000..8ca2119e --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * + * @author andrea + */ +public class CloseEvent extends MessagingEvent { + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java new file mode 100644 index 00000000..6fcdc55b --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java @@ -0,0 +1,10 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + *Event used to push work request on the SimpleMessaging. + * + * @author andrea + */ +public abstract class MessagingEvent { + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java new file mode 100644 index 00000000..355a56a3 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -0,0 +1,37 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * + * @author andrea + */ +public class PublishEvent extends MessagingEvent { + String m_topic; + QOSType m_qos; + byte[] m_message; + boolean m_retain; + + public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain) { + m_topic = topic; + m_qos = qos; + m_message = message; + m_retain = retain; + } + + public String getTopic() { + return m_topic; + } + + public QOSType getQos() { + return m_qos; + } + + public byte[] getMessage() { + return m_message; + } + + public boolean isRetain() { + return m_retain; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java new file mode 100644 index 00000000..e2832185 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java @@ -0,0 +1,19 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * + * @author andrea + */ +public class RemoveAllSubscriptionsEvent extends MessagingEvent { + + String m_clientID; + + public RemoveAllSubscriptionsEvent(String clientID) { + m_clientID = clientID; + } + + public String getClientID() { + return m_clientID; + } + +} 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 new file mode 100644 index 00000000..334e7e48 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java @@ -0,0 +1,21 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.dna.mqtt.moquette.messaging.spi.impl.Subscription; + +/** + * + * @author andrea + */ +public class SubscribeEvent extends MessagingEvent { + + Subscription m_subscription; + + public SubscribeEvent(Subscription subscription) { + m_subscription = subscription; + } + + public Subscription getSubscription() { + return m_subscription; + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java new file mode 100644 index 00000000..ab752088 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/UnsubscribeEvent.java @@ -0,0 +1,25 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * + * @author andrea + */ +public class UnsubscribeEvent extends MessagingEvent { + + String m_topic; + String m_clientID; + + public UnsubscribeEvent(String topic, String clientID) { + m_topic = topic; + m_clientID = clientID; + } + + public String getTopic() { + return m_topic; + } + + public String getClientID() { + return m_clientID; + } + +} 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 c28236fd..644e64d4 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,6 +3,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.logging.Level; import org.apache.mina.core.service.IoAcceptor; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; @@ -44,6 +45,7 @@ public class Server { File.separator + "moquette_store.hawtdb"; private IoAcceptor m_acceptor; SimpleMessaging messaging; + Thread messagingEventLoop; public static void main(String[] args) throws IOException { new Server().startServer(); @@ -77,7 +79,9 @@ public class Server { //TODO fix this hugly wiring handler.setMessaging(messaging); messaging.setNotifier(handler); - + messagingEventLoop = new Thread(messaging); + messagingEventLoop.setName("Event Loop" + System.currentTimeMillis()); + messagingEventLoop.start(); m_acceptor.setHandler(handler); ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); @@ -100,6 +104,12 @@ public class Server { LOG.info("Server stopping..."); messaging.close(); + messagingEventLoop.interrupt(); + try { + messagingEventLoop.join(); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } for(IoSession session: m_acceptor.getManagedSessions().values()) { if(session.isConnected() && !session.isClosing()){ From d34201d0d9ff0740d2d16be876e36016c48e017c Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:48:54 +0200 Subject: [PATCH 05/10] Fixed tests --- .../spi/impl/SimpleMessagingTest.java | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) 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 b90a8862..b41b8dde 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,11 +1,16 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import org.dna.mqtt.moquette.messaging.spi.INotifier; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentCaptor; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; import static org.mockito.Mockito.*; /** @@ -26,7 +31,8 @@ public class SimpleMessagingTest { @Test public void testSubscribe() { //Exercise - messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); + messaging.processSubscribe(evt); //Verify Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); @@ -35,10 +41,11 @@ public class SimpleMessagingTest { @Test public void testDoubleSubscribe() { - messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); + messaging.processSubscribe(evt); //Exercise - messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); + messaging.processSubscribe(evt); //Verify Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); @@ -47,12 +54,14 @@ public class SimpleMessagingTest { @Test public void testPublish() { - messaging.subscribe(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE); + SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); + messaging.processSubscribe(evt); INotifier notifier = mock(INotifier.class); messaging.setNotifier(notifier); //Exercise - messaging.publish(FAKE_TOPIC, "Hello".getBytes(), QOSType.MOST_ONE, false); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false); + messaging.processPublish(pubEvt); //Verify ArgumentCaptor argument = ArgumentCaptor.forClass(byte[].class); From a8773c3331aaf986986512ad86ed648bb2cabc1e Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 27 Aug 2012 11:49:30 +0200 Subject: [PATCH 06/10] Fixed imports --- .../mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java | 3 --- 1 file changed, 3 deletions(-) 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 b41b8dde..f1c91e5d 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 @@ -4,13 +4,10 @@ import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentCaptor; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.eq; import static org.mockito.Mockito.*; /** From f2220e664ad5fb2d112579b84d84ecda4cbd0f48 Mon Sep 17 00:00:00 2001 From: andrea Date: Tue, 28 Aug 2012 11:42:34 +0200 Subject: [PATCH 07/10] Removed commented code from client --- .../org/dna/mqtt/moquette/client/Client.java | 44 ------------------- 1 file changed, 44 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 8dc06cf6..87e56eb6 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java @@ -266,51 +266,8 @@ public final class Client { // m_inflightIDs.add(messageID); register(topic, publishCallback); -// boolean unlocked = false; //true when a SUBACK is received try { manageSendQoS1(msg); -// for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) { -// if (retries > 0) { -// msg.setDupFlag(true); -// } -// -// WriteFuture wf = m_session.write(msg); -// try { -// wf.await(); -// } catch (InterruptedException ex) { -// LOG.error(null, ex); -// throw new SubscribeException(ex); -// } -// LOG.info("subscribe message sent"); -// -// Throwable ex = wf.getException(); -// if (ex != null) { -// throw new SubscribeException(ex); -// } -// -// //wait for the SubAck -// m_subscribeBarrier = new CountDownLatch(1); -// -// //suspend until the server respond with CONN_ACK -// try { -// LOG.info("subscribe waiting for suback"); -// unlocked = m_subscribeBarrier.await(SUBACK_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize -// } catch (InterruptedException iex) { -// throw new SubscribeException(iex); -// } -// } -// -// //if not arrive into certain limit, raise an error -// if (!unlocked) { -// throw new SubscribeException(String.format("Server doesn't replyed with a SUB_ACK after %d replies", RETRIES_QOS_GT0)); -// } else { -// //check if message ID match -// if (m_receivedSubAckMessageID != messageID) { -// throw new SubscribeException(String.format("Server replyed with " -// + "a broken MessageID in SUB_ACK, expected %d but received %d", -// messageID, m_receivedSubAckMessageID)); -// } -// } } catch(Throwable ex) { //in case errors arise, remove the registration because the subscription // hasn't get well @@ -318,7 +275,6 @@ public final class Client { throw new MQTTException(ex); } -// register(topic, publishCallback); updatePinger(); } From bb3e8170d5719818e65d13bdb829022706bba731 Mon Sep 17 00:00:00 2001 From: andrea Date: Wed, 29 Aug 2012 14:40:37 +0200 Subject: [PATCH 08/10] Added log --- parser/src/main/java/org/dna/mqtt/moquette/client/Client.java | 1 + 1 file changed, 1 insertion(+) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 87e56eb6..10f5bf57 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java @@ -311,6 +311,7 @@ public final class Client { int messageID = msg.getMessageID(); boolean unlocked = false; for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) { + LOG.debug("manageSendQoS1 retry " + retries); if (retries > 0) { msg.setDupFlag(true); } From 323811adb2c069b6bf44ea78952f632d94e47b01 Mon Sep 17 00:00:00 2001 From: andrea Date: Wed, 29 Aug 2012 21:20:36 +0200 Subject: [PATCH 09/10] Added outbound event queue to be handled by Notifier stage --- .../moquette/messaging/spi/IMessaging.java | 3 + .../moquette/messaging/spi/INotifier.java | 3 + .../messaging/spi/impl/SimpleMessaging.java | 62 +++++++++++++------ .../spi/impl/events/DisconnectEvent.java | 22 +++++++ .../spi/impl/events/NotifyEvent.java | 46 ++++++++++++++ .../dna/mqtt/moquette/server/MQTTHandler.java | 48 ++++++++++---- .../dna/mqtt/moquette/server/Notifier.java | 48 ++++++++++++++ .../org/dna/mqtt/moquette/server/Server.java | 21 ++++++- parser/src/main/resources/log4j.properties | 8 ++- .../spi/impl/SimpleMessagingTest.java | 26 +++++--- .../server/ServerIntegrationTest.java | 9 ++- 11 files changed, 249 insertions(+), 47 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java 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 0754c059..bf5623ee 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 @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi; +import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -35,4 +36,6 @@ public interface IMessaging { * doesn't reply any error */ void unsubscribe(String topic, String clientID); + + void disconnect(IoSession session); } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java index b9c4be98..13685f38 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi; +import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -9,5 +10,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; public interface INotifier { public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained); + + public void disconnect(IoSession session); } 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 f58b759a..75d4c254 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 @@ -1,7 +1,5 @@ package org.dna.mqtt.moquette.messaging.spi.impl; -import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import java.io.File; import java.io.IOException; import java.io.Serializable; @@ -10,12 +8,16 @@ import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; import org.dna.mqtt.moquette.messaging.spi.impl.events.CloseEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent; +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.RemoveAllSubscriptionsEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.UnsubscribeEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; @@ -58,13 +60,13 @@ public class SimpleMessaging implements IMessaging, Runnable { private SubscriptionsStore subscriptions = new SubscriptionsStore(); private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); + private BlockingQueue m_outboundQueue = new LinkedBlockingQueue(); - private INotifier m_notifier; +// private INotifier m_notifier; private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; private SortedIndex m_retainedStore; - public SimpleMessaging() { String storeFile = Server.STORAGE_FILE_PATH; @@ -98,9 +100,6 @@ public class SimpleMessaging implements IMessaging, Runnable { eventLoop(); } - public void setNotifier(INotifier notifier) { - m_notifier = notifier; - } public void publish(String topic, byte[] message, QOSType qos, boolean retain) { //TODO for this moment no qos > 0 trivial implementation @@ -116,6 +115,7 @@ public class SimpleMessaging implements IMessaging, Runnable { public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); try { + LOG.debug("subscribe invoked for topic: " + topic); m_inboundQueue.put(new SubscribeEvent(newSubscription)); } catch (InterruptedException ex) { LOG.error(null, ex); @@ -132,6 +132,15 @@ public class SimpleMessaging implements IMessaging, Runnable { } + public void disconnect(IoSession session) { + try { + m_inboundQueue.put(new DisconnectEvent(session)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } + + /** * Verify if the 2 topics matching respecting the rules of MQTT Appendix A */ @@ -183,7 +192,11 @@ public class SimpleMessaging implements IMessaging, Runnable { m_inboundQueue.put(new CloseEvent()); } catch (InterruptedException ex) { LOG.error(null, ex); - } + } + } + + public BlockingQueue getNotifyEventQueue() { + return m_outboundQueue; } @@ -203,6 +216,8 @@ public class SimpleMessaging implements IMessaging, Runnable { processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt); } else if (evt instanceof CloseEvent) { processClose(); + } else if (evt instanceof DisconnectEvent) { + m_outboundQueue.put(evt); } } catch (InterruptedException ex) { processClose(); @@ -213,13 +228,17 @@ public class SimpleMessaging implements IMessaging, Runnable { protected void processPublish(PublishEvent evt) { LOG.debug("processPublish invoked"); - String topic = evt.getTopic(); - QOSType qos = evt.getQos(); - byte[] message = evt.getMessage(); + final String topic = evt.getTopic(); + final QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); boolean retain = evt.isRetain(); - for (Subscription sub : subscriptions.matches(topic)) { - m_notifier.notify(sub.clientId, topic, qos, message, false); + for (final Subscription sub : subscriptions.matches(topic)) { + try { + m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } } if (retain) { if (message.length == 0) { @@ -233,7 +252,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } - protected void processSubscribe(SubscribeEvent evt) { + protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ { LOG.debug("processSubscribe invoked"); Subscription newSubscription = evt.getSubscription(); String topic = newSubscription.getTopic(); @@ -241,13 +260,18 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription - LOG.debug("Scanning all retained messages..."); + LOG.debug("Scanning all retained messages, presents are " + m_retainedStore.size()); for (Map.Entry entry : m_retainedStore) { StoredMessage storedMsg = entry.getValue(); if (matchTopics(entry.getKey(), topic)) { - //fire the as retained the message - m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), - storedMsg.getPayload(), true); + try { + //fire the as retained the message + LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey()); + m_outboundQueue.put(new NotifyEvent(newSubscription.clientId, + topic, storedMsg.getQos(), storedMsg.getPayload(), true)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } } } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java new file mode 100644 index 00000000..bf8ecd27 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java @@ -0,0 +1,22 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.apache.mina.core.session.IoSession; + +/** + * + * @author andrea + */ +public class DisconnectEvent extends MessagingEvent { + + IoSession m_session; + + public DisconnectEvent(IoSession session) { + m_session = session; + } + + public IoSession getSession() { + return m_session; + } + + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java new file mode 100644 index 00000000..4fe6dc2e --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java @@ -0,0 +1,46 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * + * @author andrea + */ +public class NotifyEvent extends MessagingEvent { + + private String m_clientId; + private String m_topic; + private QOSType m_qos; + private byte[] m_message; + private boolean m_retaned; + + + public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained) { + m_clientId = clientId; + m_topic = topic; + m_qos = qos; + m_message = message; + m_retaned = retained; + } + + public String getClientId() { + return m_clientId; + } + + public String getTopic() { + return m_topic; + } + + public QOSType getQos() { + return m_qos; + } + + public byte[] getMessage() { + return m_message; + } + + public boolean isRetained() { + return m_retaned; + } + +} 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 c1736382..367cbef0 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 @@ -63,7 +63,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { break; case DISCONNECT: handleDisconnect(session, (DisconnectMessage) msg); - + break; } } catch (Exception ex) { LOG.error("Bad error in processing the message", ex); @@ -152,7 +152,7 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } protected void handleSubscribe(IoSession session, SubscribeMessage msg) { - LOG.info("registering the subscriptions"); + LOG.debug("handleSubscribe, registering the subscriptions"); for (SubscribeMessage.Couple req : msg.subscriptions()) { m_messaging.subscribe((String) session.getAttribute(ATTR_CLIENTID), req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]); @@ -191,12 +191,12 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) { String clientID = (String) session.getAttribute(ATTR_CLIENTID); //remove from clientIDs - m_clientIDsLock.lock(); - try { - m_clientIDs.remove(clientID); - } finally { - m_clientIDsLock.unlock(); - } +// m_clientIDsLock.lock(); +// try { +// m_clientIDs.remove(clientID); +// } finally { +// m_clientIDsLock.unlock(); +// } boolean cleanSession = (Boolean) session.getAttribute("cleanSession"); if (cleanSession) { //cleanup topic subscriptions @@ -204,7 +204,8 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } //close the TCP connection - session.close(true); + //session.close(true); + m_messaging.disconnect(session); } @Override @@ -226,9 +227,30 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { pubMessage.setTopicName(topic); pubMessage.setQos(qOSType); pubMessage.setPayload(payload); - assert m_clientIDs != null; - assert m_clientIDs.get(clientId) != null; - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); - m_clientIDs.get(clientId).getSession().write(pubMessage); + + LOG.debug("notify invoked"); + m_clientIDsLock.lock(); + try { + assert m_clientIDs != null; + LOG.debug("clientIDs are " + m_clientIDs); + assert m_clientIDs.get(clientId) != null; + LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + m_clientIDs.get(clientId).getSession().write(pubMessage); + }catch(Throwable t) { + LOG.error(null, t); + } finally { + m_clientIDsLock.unlock(); + } + } + + public void disconnect(IoSession session) { + String clientID = (String) session.getAttribute(ATTR_CLIENTID); + m_clientIDsLock.lock(); + try { + m_clientIDs.remove(clientID); + } finally { + m_clientIDsLock.unlock(); + } + session.close(true); } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java new file mode 100644 index 00000000..676184b9 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java @@ -0,0 +1,48 @@ +package org.dna.mqtt.moquette.server; + +import java.util.concurrent.BlockingQueue; +import org.dna.mqtt.moquette.messaging.spi.INotifier; +import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +class Notifier implements Runnable { + + private static final Logger LOG = LoggerFactory.getLogger(Notifier.class); + + BlockingQueue m_queue; + INotifier m_notifier; + + Notifier(BlockingQueue queue, INotifier notifier) { + m_queue = queue; + m_notifier = notifier; + } + + public void run() { + LOG.debug("Notifier started"); + boolean interrupted = false; + while (!interrupted) { + try { + MessagingEvent evt = m_queue.take(); + LOG.debug("Notifing event " + evt); + if (evt instanceof NotifyEvent) { + NotifyEvent evtN = (NotifyEvent) evt; + m_notifier.notify(evtN.getClientId(), evtN.getTopic(), evtN.getQos(), evtN.getMessage(), evtN.isRetained()); + } else if (evt instanceof DisconnectEvent) { + DisconnectEvent evtD = (DisconnectEvent) evt; + m_notifier.disconnect(evtD.getSession()); + } + } catch (InterruptedException ex) { + interrupted = true; + } + } + LOG.debug("Notifier stopped"); + } + +} 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 644e64d4..91b2a64d 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,7 +3,9 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.logging.Level; +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.session.IdleStatus; import org.apache.mina.core.session.IoSession; @@ -12,6 +14,7 @@ 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.ConnAckEncoder; import org.dna.mqtt.moquette.proto.ConnectDecoder; import org.dna.mqtt.moquette.proto.DisconnectDecoder; @@ -43,9 +46,13 @@ public class Server { public static final int DEFAULT_CONNECT_TIMEOUT = 10; public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + File.separator + "moquette_store.hawtdb"; + + //TODO one notifier per outbound queue, else no serialization of message flow is granted!! + private static final int NOTIFIER_POOL_SIZE = 1; private IoAcceptor m_acceptor; SimpleMessaging messaging; Thread messagingEventLoop; + private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/; public static void main(String[] args) throws IOException { new Server().startServer(); @@ -78,11 +85,18 @@ public class Server { messaging = new SimpleMessaging(); //TODO fix this hugly wiring handler.setMessaging(messaging); - messaging.setNotifier(handler); +// messaging.setNotifier(handler); messagingEventLoop = new Thread(messaging); messagingEventLoop.setName("Event Loop" + System.currentTimeMillis()); messagingEventLoop.start(); + m_notifierPool = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE); + + BlockingQueue notifyQueue = messaging.getNotifyEventQueue(); + for (int i = 0; i < NOTIFIER_POOL_SIZE; i++) { + m_notifierPool.submit(new Notifier(notifyQueue, handler)); + } + m_acceptor.setHandler(handler); ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); @@ -105,12 +119,15 @@ public class Server { messaging.close(); messagingEventLoop.interrupt(); + LOG.info("shutting down evet loop"); try { messagingEventLoop.join(); } catch (InterruptedException ex) { LOG.error(null, ex); } + m_notifierPool.shutdown(); + for(IoSession session: m_acceptor.getManagedSessions().values()) { if(session.isConnected() && !session.isClosing()){ session.close(false); diff --git a/parser/src/main/resources/log4j.properties b/parser/src/main/resources/log4j.properties index 68b844a6..8b5ada77 100644 --- a/parser/src/main/resources/log4j.properties +++ b/parser/src/main/resources/log4j.properties @@ -1,5 +1,11 @@ # Set root logger level to DEBUG and its only appender to A1. -log4j.rootLogger=DEBUG, A1 +log4j.rootLogger=ERROR, A1 +#log4j.rootLogger=DEBUG, A1 +#log4j.rootLogger=WARN, A1 +#log4j.logger.org.dna.mqtt.moquette.client=WARN, A1 +#log4j.logger.org.mina=WARN, A1 +#log4j.logger.org.dna.mqtt.moquette.server=DEBUG, A1 +#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG, A1 # A1 is set to be a ConsoleAppender. log4j.appender.A1=org.apache.log4j.ConsoleAppender 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 f1c91e5d..fe19a9af 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,14 +1,14 @@ package org.dna.mqtt.moquette.messaging.spi.impl; -import org.dna.mqtt.moquette.messaging.spi.INotifier; +import java.util.concurrent.BlockingQueue; +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.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; -import org.mockito.ArgumentCaptor; -import static org.mockito.Mockito.*; /** * @@ -50,20 +50,28 @@ public class SimpleMessagingTest { } @Test - public void testPublish() { + public void testPublish() throws InterruptedException { SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE)); messaging.processSubscribe(evt); - INotifier notifier = mock(INotifier.class); - messaging.setNotifier(notifier); +// INotifier notifier = mock(INotifier.class); +// messaging.setNotifier(notifier); //Exercise PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false); messaging.processPublish(pubEvt); //Verify - ArgumentCaptor argument = ArgumentCaptor.forClass(byte[].class); - verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture(), eq(false)); - assertEquals("Hello", new String(argument.getValue())); + BlockingQueue queue = messaging.getNotifyEventQueue(); + MessagingEvent msgEvt = queue.take(); + assertTrue(msgEvt instanceof NotifyEvent); + NotifyEvent notifyEvt = (NotifyEvent) msgEvt; + assertEquals(FAKE_CLIENT_ID, notifyEvt.getClientId()); + assertEquals(FAKE_TOPIC, notifyEvt.getTopic()); + assertFalse(notifyEvt.isRetained()); + +// ArgumentCaptor argument = ArgumentCaptor.forClass(byte[].class); +// verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture(), eq(false)); + assertEquals("Hello", new String(notifyEvt.getMessage())); } @Test diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 49020886..01b952cd 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -9,6 +9,7 @@ import org.dna.mqtt.moquette.client.IPublishCallback; import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; /** @@ -203,7 +204,8 @@ public class ServerIntegrationTest { @Test - public void testRetain_maintainMessage_againstClientDestruction() { + public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException { + final CountDownLatch barrier = new CountDownLatch(1); Client client = new Client("localhost", Server.PORT, "CLID_123"); client.connect(); @@ -218,11 +220,13 @@ public class ServerIntegrationTest { public void published(String topic, byte[] message) { received = true; + barrier.countDown(); } }); client.close(); //Verify + barrier.await(1, TimeUnit.SECONDS); assertTrue(received); //TearDown @@ -230,7 +234,7 @@ public class ServerIntegrationTest { } - @Test + @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); @@ -308,7 +312,6 @@ public class ServerIntegrationTest { client.close(); client.shutdown(); - } } From 87791963b696d3f881ba2a7d7f1df15be14cb34d Mon Sep 17 00:00:00 2001 From: andrea Date: Wed, 29 Aug 2012 21:46:47 +0200 Subject: [PATCH 10/10] Defined constant for client's number of messages published --- .../src/main/java/org/dna/mqtt/bechnmark/Consumer.java | 4 +--- .../src/main/java/org/dna/mqtt/bechnmark/Producer.java | 2 +- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java index d2670423..2c9f1a90 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java @@ -15,8 +15,6 @@ import org.slf4j.LoggerFactory; */ public class Consumer implements Runnable { - static final int PUB_LOOP = 10000; - private static final Logger LOG = LoggerFactory.getLogger(Consumer.class); private String m_clientID; @@ -54,7 +52,7 @@ public class Consumer implements Runnable { } Message message = null; - for (int i = 0; i < PUB_LOOP; i++) { + for (int i = 0; i < Producer.PUB_LOOP; i++) { try { message = connection.receive(); } catch (Exception ex) { diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java index 4a82dfdd..a0e879dc 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java @@ -19,7 +19,7 @@ public class Producer implements Runnable { private String m_clientID; - static final int PUB_LOOP = 10000; + public static final int PUB_LOOP = 1000000; public Producer(String clientID) { m_clientID = clientID;