From 9276ea8c98617e818cc335f690232308e367bff0 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 12:58:40 +0200 Subject: [PATCH] Removed back notifier loop --- .../messaging/spi/impl/SimpleMessaging.java | 46 +++++++++++++------ .../org/dna/mqtt/moquette/server/Server.java | 12 ++--- .../spi/impl/SimpleMessagingTest.java | 27 +++++++---- 3 files changed, 54 insertions(+), 31 deletions(-) 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 ff5d84dc..ab45bd82 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 @@ -12,6 +12,7 @@ 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.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; @@ -55,9 +56,9 @@ public class SimpleMessaging implements IMessaging, Runnable { private SubscriptionsStore subscriptions = new SubscriptionsStore(); private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); - private BlockingQueue m_outboundQueue = 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; @@ -115,6 +116,10 @@ public class SimpleMessaging implements IMessaging, Runnable { m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); } + + public void setNotifier(INotifier notifier) { + m_notifier= notifier; + } public void run() { eventLoop(); @@ -233,9 +238,9 @@ public class SimpleMessaging implements IMessaging, Runnable { } } - public BlockingQueue getNotifyEventQueue() { + /*public BlockingQueue getNotifyEventQueue() { return m_outboundQueue; - } + }*/ protected void eventLoop() { @@ -288,7 +293,8 @@ public class SimpleMessaging implements IMessaging, Runnable { for (final Subscription sub : subscriptions.matches(topic)) { if (qos == QOSType.MOST_ONE) { //QoS 0 - m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); + m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); +// m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it @@ -303,13 +309,17 @@ public class SimpleMessaging implements IMessaging, Runnable { storedEvents.add(evt); m_persistentMessageStore.put(clientID, storedEvents); } - m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); + m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); +// m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); } } if (cleanEvt != null) { - m_outboundQueue.put(cleanEvt); - m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID())); + refill(cleanEvt); + m_notifier.sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); + +// m_outboundQueue.put(cleanEvt); +// m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID())); } if (retain) { @@ -336,14 +346,16 @@ public class SimpleMessaging implements IMessaging, Runnable { for (Map.Entry entry : m_retainedStore) { StoredMessage storedMsg = entry.getValue(); if (matchTopics(entry.getKey(), topic)) { - try { +// 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, + m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(), storedMsg.getPayload(), true)); - } catch (InterruptedException ex) { - LOG.error(null, ex); - } +// m_outboundQueue.put(new NotifyEvent(newSubscription.clientId, +// topic, storedMsg.getQos(), storedMsg.getPayload(), true)); +// } catch (InterruptedException ex) { +// LOG.error(null, ex); +// } } } } @@ -362,7 +374,8 @@ public class SimpleMessaging implements IMessaging, Runnable { } private void processDisconnect(DisconnectEvent evt) throws InterruptedException { - m_outboundQueue.put(evt); +// m_outboundQueue.put(evt); + m_notifier.disconnect(evt.getSession()); //de-activate the subscriptions for this ClientID String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); @@ -385,8 +398,11 @@ public class SimpleMessaging implements IMessaging, Runnable { } for (PublishEvent pubEvt : publishedEvents) { - m_outboundQueue.put(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + m_notifier.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID())); + +// m_outboundQueue.put(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), +// pubEvt.getMessage(), false, pubEvt.getMessageID())); } } } 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 ca9cb9e8..9e1a01e7 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 @@ -33,11 +33,11 @@ public class Server { 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 static final int NOTIFIER_POOL_SIZE = 1; private IoAcceptor m_acceptor; SimpleMessaging messaging; Thread messagingEventLoop; - private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/; + //private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/; public static void main(String[] args) throws IOException { new Server().startServer(); @@ -71,18 +71,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); + /*m_notifierPool = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE); BlockingQueue notifyQueue = messaging.getNotifyEventQueue(); for (int i = 0; i < NOTIFIER_POOL_SIZE; i++) { Notifier notifier = new Notifier(notifyQueue, handler, messaging); m_notifierPool.submit(notifier); - } + }*/ m_acceptor.setHandler(handler); ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); @@ -113,7 +113,7 @@ public class Server { LOG.error(null, ex); } - m_notifierPool.shutdown(); + /*m_notifierPool.shutdown();*/ for(IoSession session: m_acceptor.getManagedSessions().values()) { if(session.isConnected() && !session.isClosing()){ 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 404aebf7..d4a49c9e 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,6 +1,8 @@ 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; @@ -9,6 +11,8 @@ 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.*; /** * @@ -53,25 +57,28 @@ public class SimpleMessagingTest { public void testPublish() throws InterruptedException { SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false)); 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, "FakeCLI", null); messaging.processPublish(pubEvt); //Verify - 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()); +// 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())); +// assertEquals("Hello", new String(notifyEvt.getMessage())); + + ArgumentCaptor argument = ArgumentCaptor.forClass(NotifyEvent.class); + verify(notifier).notify(argument.capture()); } @Test