anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed a ping pong beteen the ProtocolProcessor and SimpleMessaging, avoid call method, send Event, receive Event e callback a method into ProcotolProcessor

Browse Source
This commit is contained in:
andrea
2013-08-30 19:05:09 +02:00
parent ff0139d809
commit 2c11cade06
3 changed files with 39 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+1 -5
View File
@@ -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);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+21 -4
View File
@@ -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<PublishEvent> 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();
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+17 -17
View File
@@ -92,11 +92,11 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
// 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<ValueEvent> {
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<ValueEvent> {
m_executor.shutdown();
}
private void processRepublish(RepublishEvent evt) throws InterruptedException {
LOG.debug("processRepublish invoked");
List<PublishEvent> 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<PublishEvent> 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()));
// }
// }
}