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 ab45bd82..716db1f0 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 @@ -56,8 +56,7 @@ 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 MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; @@ -238,10 +237,6 @@ public class SimpleMessaging implements IMessaging, Runnable { } } - /*public BlockingQueue getNotifyEventQueue() { - return m_outboundQueue; - }*/ - protected void eventLoop() { LOG.debug("Started event loop"); @@ -294,7 +289,6 @@ public class SimpleMessaging implements IMessaging, Runnable { if (qos == QOSType.MOST_ONE) { //QoS 0 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 @@ -310,16 +304,12 @@ public class SimpleMessaging implements IMessaging, Runnable { m_persistentMessageStore.put(clientID, storedEvents); } 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) { 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) { @@ -334,7 +324,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } - protected void processSubscribe(SubscribeEvent evt)/* throws InterruptedException*/ { + protected void processSubscribe(SubscribeEvent evt) { LOG.debug("processSubscribe invoked"); Subscription newSubscription = evt.getSubscription(); String topic = newSubscription.getTopic(); @@ -346,16 +336,10 @@ public class SimpleMessaging implements IMessaging, Runnable { for (Map.Entry entry : m_retainedStore) { StoredMessage storedMsg = entry.getValue(); if (matchTopics(entry.getKey(), topic)) { -// try { - //fire the as retained the message - LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey()); - m_notifier.notify(new NotifyEvent(newSubscription.clientId, - topic, storedMsg.getQos(), storedMsg.getPayload(), true)); -// m_outboundQueue.put(new NotifyEvent(newSubscription.clientId, -// topic, storedMsg.getQos(), storedMsg.getPayload(), true)); -// } catch (InterruptedException ex) { -// LOG.error(null, ex); -// } + //fire the as retained the message + LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey()); + m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(), + storedMsg.getPayload(), true)); } } } @@ -374,7 +358,6 @@ public class SimpleMessaging implements IMessaging, Runnable { } private void processDisconnect(DisconnectEvent evt) throws InterruptedException { -// m_outboundQueue.put(evt); m_notifier.disconnect(evt.getSession()); //de-activate the subscriptions for this ClientID @@ -400,9 +383,6 @@ public class SimpleMessaging implements IMessaging, Runnable { for (PublishEvent pubEvt : publishedEvents) { 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/Notifier.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java deleted file mode 100644 index ed331543..00000000 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java +++ /dev/null @@ -1,54 +0,0 @@ -package org.dna.mqtt.moquette.server; - -import java.util.concurrent.BlockingQueue; - -import org.dna.mqtt.moquette.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.messaging.spi.INotifier; -import org.dna.mqtt.moquette.messaging.spi.impl.events.*; -import org.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; - IMessaging m_messaging; - - Notifier(BlockingQueue queue, INotifier notifier, IMessaging messaging) { - m_queue = queue; - m_notifier = notifier; - m_messaging = messaging; - } - - 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); - } else if (evt instanceof DisconnectEvent) { - DisconnectEvent evtD = (DisconnectEvent) evt; - m_notifier.disconnect(evtD.getSession()); - } else if (evt instanceof CleanInFlightEvent) { - m_messaging.refill(evt); - } else if (evt instanceof PubAckEvent) { - m_notifier.sendPubAck((PubAckEvent) evt); - } - } 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 9e1a01e7..7392d2f9 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 @@ -76,14 +76,6 @@ public class Server { 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++) { - Notifier notifier = new Notifier(notifyQueue, handler, messaging); - m_notifierPool.submit(notifier); - }*/ - m_acceptor.setHandler(handler); ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); 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 d4a49c9e..924427c7 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 @@ -65,18 +65,6 @@ public class SimpleMessagingTest { 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()); - -// 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())); - ArgumentCaptor argument = ArgumentCaptor.forClass(NotifyEvent.class); verify(notifier).notify(argument.capture()); }