From 2c11cade06323ca7b8ee1be6202e32290cb9ec5a Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 19:05:09 +0200 Subject: [PATCH] Removed a ping pong beteen the ProtocolProcessor and SimpleMessaging, avoid call method, send Event, receive Event e callback a method into ProcotolProcessor --- .../moquette/messaging/spi/IMessaging.java | 6 +--- .../messaging/spi/impl/ProtocolProcessor.java | 25 +++++++++++--- .../messaging/spi/impl/SimpleMessaging.java | 34 +++++++++---------- 3 files changed, 39 insertions(+), 26 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 3cad0278..eeccae6b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -1,11 +1,7 @@ package org.dna.mqtt.moquette.messaging.spi; import org.apache.mina.core.session.IoSession; -import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; /** * Interface to the underling messaging system used to publish, subscribe. @@ -18,7 +14,7 @@ public interface IMessaging { void disconnect(IoSession session); - void republishStored(String clientID); +// void republishStored(String clientID); void handleProtocolMessage(IoSession session, AbstractMessage msg); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 8e075ab9..45a02907 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -11,6 +11,7 @@ import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.PubCompMessage; @@ -130,14 +131,30 @@ class ProtocolProcessor { //remove all prev subscriptions //cleanup topic subscriptions processRemoveAllSubscriptions(msg.getClientID()); - } else { - //force the republish of stored QoS1 and QoS2 - m_messaging.republishStored(msg.getClientID()); } ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); session.write(okResp); + + if (!msg.isCleanSession()) { + //force the republish of stored QoS1 and QoS2 + republishStored(msg.getClientID()); + } + } + + private void republishStored(String clientID) { + LOG.debug("republishStored invoked"); + List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); + if (publishedEvents == null) { + LOG.debug("republishStored, no stored publish events"); + return; + } + + for (PublishEvent pubEvt : publishedEvents) { + notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), + pubEvt.getMessage(), false, pubEvt.getMessageID())); + } } private void processRemoveAllSubscriptions(String clientID) { @@ -211,7 +228,7 @@ class ProtocolProcessor { } } - void notify(NotifyEvent evt) { + private void notify(NotifyEvent evt) { LOG.debug("notify invoked with event " + evt); String clientId = evt.getClientId(); PublishMessage pubMessage = new PublishMessage(); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 07fe461f..1d106873 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -92,11 +92,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { // disruptorPublish(evt); // } - public void republishStored(String clientID) { + /*public void republishStored(String clientID) { //create the event to push LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID); disruptorPublish(new RepublishEvent(clientID)); - } + }*/ public void handleProtocolMessage(IoSession session, AbstractMessage msg) { disruptorPublish(new ProtocolEvent(session, msg)); @@ -117,9 +117,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); m_processor.processDisconnect(disEvt.getSession(), clientID, false); - } else if (evt instanceof RepublishEvent) { + } /*else if (evt instanceof RepublishEvent) { processRepublish((RepublishEvent) evt); - } else if (evt instanceof ProtocolEvent) { + } */else if (evt instanceof ProtocolEvent) { IoSession session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); if (message instanceof ConnectMessage) { @@ -190,17 +190,17 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_executor.shutdown(); } - private void processRepublish(RepublishEvent evt) throws InterruptedException { - LOG.debug("processRepublish invoked"); - List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); - if (publishedEvents == null) { - LOG.debug("processRepublish, no stored publish events"); - return; - } - - for (PublishEvent pubEvt : publishedEvents) { - m_processor.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), - pubEvt.getMessage(), false, pubEvt.getMessageID())); - } - } +// private void processRepublish(RepublishEvent evt) throws InterruptedException { +// LOG.debug("processRepublish invoked"); +// List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); +// if (publishedEvents == null) { +// LOG.debug("processRepublish, no stored publish events"); +// return; +// } +// +// for (PublishEvent pubEvt : publishedEvents) { +// m_processor.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), +// pubEvt.getMessage(), false, pubEvt.getMessageID())); +// } +// } }