Removed clean session flag from subscription and moved in the correct place (ClientSession)

This commit is contained in:
andsel
2015-12-06 19:55:04 +01:00
parent 80569d5dfe
commit 5a1cd97b71
8 changed files with 56 additions and 64 deletions
@@ -26,7 +26,6 @@ import io.moquette.spi.ClientSession;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.impl.security.IAuthenticator;
import io.moquette.spi.impl.security.IAuthorizator;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
@@ -384,8 +383,7 @@ public class ProtocolProcessor {
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
//TODO isCleanSession MUST be on the targetSession not the Subscription
if (!sub.isCleanSession() && !targetSession.isActive()) {
if (!targetSession.isCleanSession() && !targetSession.isActive()) {
//store the message in targetSession queue to deliver
targetSession.enqueueToDeliver(guid);
} else {
@@ -594,7 +592,6 @@ public class ProtocolProcessor {
public void processSubscribe(ServerChannel session, SubscribeMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
@@ -605,7 +602,7 @@ public class ProtocolProcessor {
List<Subscription> newSubscriptions = new ArrayList<>();
for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos());
Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession);
Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos);
//boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter());
boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription);
ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE);
@@ -31,21 +31,18 @@ public class Subscription implements Serializable {
QOSType requestedQos; //max QoS acceptable
String clientId;
String topicFilter;
boolean cleanSession;
boolean active = true;
public Subscription(String clientId, String topicFilter, QOSType requestedQos, boolean cleanSession) {
public Subscription(String clientId, String topicFilter, QOSType requestedQos) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topicFilter = topicFilter;
this.cleanSession = cleanSession;
}
public Subscription(Subscription orig) {
this.requestedQos = orig.requestedQos;
this.clientId = orig.clientId;
this.topicFilter = orig.topicFilter;
this.cleanSession = orig.cleanSession;
this.active = orig.active;
}
@@ -61,14 +58,6 @@ public class Subscription implements Serializable {
return topicFilter;
}
/**
* @deprecated this concept must be moved inside ClientSession and not subscription.
* */
@Deprecated
public boolean isCleanSession() {
return this.cleanSession;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;