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

Changed some methods visibility

Browse Source
This commit is contained in:
andrea
2013-08-30 18:55:11 +02:00
parent 2b7858de7c
commit ff0139d809
2 changed files with 12 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+7 -3
View File
@@ -140,7 +140,7 @@ class ProtocolProcessor {
session.write(okResp);
}
void processRemoveAllSubscriptions(String clientID) {
private void processRemoveAllSubscriptions(String clientID) {
LOG.debug("processRemoveAllSubscriptions invoked");
subscriptions.removeForClient(clientID);
@@ -185,7 +185,7 @@ class ProtocolProcessor {
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
for (final Subscription sub : subscriptions.matches(topic)) {
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS 0
@@ -316,7 +316,11 @@ class ProtocolProcessor {
m_storageService.cleanInFlight(publishKey);
}
void processDisconnect(IoSession session, String clientID) throws InterruptedException {
void processDisconnect(IoSession session, String clientID, boolean cleanSession) throws InterruptedException {
if (cleanSession) {
//cleanup topic subscriptions
processRemoveAllSubscriptions(clientID);
}
// m_notifier.disconnect(evt.getSession());
m_clientIDs.remove(clientID);
session.close(true);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+5 -18
View File
@@ -88,9 +88,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
//method used by hte Notifier to re-put an event on the inbound queue
private void refill(MessagingEvent evt) {
disruptorPublish(evt);
}
// private void refill(MessagingEvent evt) {
// disruptorPublish(evt);
// }
public void republishStored(String clientID) {
//create the event to push
@@ -102,15 +102,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
disruptorPublish(new ProtocolEvent(session, msg));
}
/**
* NOT SAFE Method, to be removed because used only in tests
*/
protected SubscriptionsStore getSubscriptions() {
return subscriptions;
}
public void stop() {
disruptorPublish(new StopEvent());
}
@@ -125,7 +116,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent disEvt = (DisconnectEvent) evt;
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
m_processor.processDisconnect(disEvt.getSession(), clientID);
m_processor.processDisconnect(disEvt.getSession(), clientID, false);
} else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
} else if (evt instanceof ProtocolEvent) {
@@ -149,14 +140,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_processor.processRemoveAllSubscriptions(clientID);
}
//close the TCP connection
//session.close(true);
m_processor.processDisconnect(session, clientID);
m_processor.processDisconnect(session, clientID, cleanSession);
} else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);