Minor change, renamed topic to topicFilter in Subscription
This commit is contained in:
+1
-1
@@ -246,7 +246,7 @@ public class HawtDBStorageService implements IStorageService {
|
||||
//TODO check the subs doesn't contain another subscription to the same topic with different
|
||||
Subscription existingSubscription = null;
|
||||
for (Subscription scanSub : subs) {
|
||||
if (newSubscription.getTopic().equals(scanSub.getTopic())) {
|
||||
if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) {
|
||||
existingSubscription = scanSub;
|
||||
break;
|
||||
}
|
||||
|
||||
+1
-1
@@ -308,7 +308,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
|
||||
|
||||
ByteBuffer message = origMessage.duplicate();
|
||||
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
|
||||
sub.getClientId(), sub.getTopic(), qos, sub.isActive());
|
||||
sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive());
|
||||
|
||||
if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) {
|
||||
//QoS 0
|
||||
|
||||
+10
-10
@@ -27,16 +27,16 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
*/
|
||||
public class Subscription implements Serializable {
|
||||
|
||||
QOSType requestedQos;
|
||||
QOSType requestedQos; //max QoS acceptable
|
||||
String clientId;
|
||||
String topic;
|
||||
String topicFilter;
|
||||
boolean cleanSession;
|
||||
boolean active = true;
|
||||
|
||||
public Subscription(String clientId, String topic, QOSType requestedQos, boolean cleanSession) {
|
||||
public Subscription(String clientId, String topicFilter, QOSType requestedQos, boolean cleanSession) {
|
||||
this.requestedQos = requestedQos;
|
||||
this.clientId = clientId;
|
||||
this.topic = topic;
|
||||
this.topicFilter = topicFilter;
|
||||
this.cleanSession = cleanSession;
|
||||
}
|
||||
|
||||
@@ -48,8 +48,8 @@ public class Subscription implements Serializable {
|
||||
return requestedQos;
|
||||
}
|
||||
|
||||
public String getTopic() {
|
||||
return topic;
|
||||
public String getTopicFilter() {
|
||||
return topicFilter;
|
||||
}
|
||||
|
||||
public boolean isCleanSession() {
|
||||
@@ -79,7 +79,7 @@ public class Subscription implements Serializable {
|
||||
if ((this.clientId == null) ? (other.clientId != null) : !this.clientId.equals(other.clientId)) {
|
||||
return false;
|
||||
}
|
||||
if ((this.topic == null) ? (other.topic != null) : !this.topic.equals(other.topic)) {
|
||||
if ((this.topicFilter == null) ? (other.topicFilter != null) : !this.topicFilter.equals(other.topicFilter)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
@@ -90,7 +90,7 @@ public class Subscription implements Serializable {
|
||||
int hash = 3;
|
||||
hash = 37 * hash + (this.requestedQos != null ? this.requestedQos.hashCode() : 0);
|
||||
hash = 37 * hash + (this.clientId != null ? this.clientId.hashCode() : 0);
|
||||
hash = 37 * hash + (this.topic != null ? this.topic.hashCode() : 0);
|
||||
hash = 37 * hash + (this.topicFilter != null ? this.topicFilter.hashCode() : 0);
|
||||
return hash;
|
||||
}
|
||||
|
||||
@@ -98,11 +98,11 @@ public class Subscription implements Serializable {
|
||||
* Trivial match method
|
||||
*/
|
||||
boolean match(String topic) {
|
||||
return this.topic.equals(topic);
|
||||
return this.topicFilter.equals(topic);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.format("[t:%s, cliID: %s, qos: %s, active: %s]", this.topic, this.clientId, this.requestedQos, this.active);
|
||||
return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientId, this.requestedQos, this.active);
|
||||
}
|
||||
}
|
||||
|
||||
+5
-5
@@ -90,7 +90,7 @@ public class SubscriptionsStore {
|
||||
}
|
||||
|
||||
for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
|
||||
LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic());
|
||||
LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter());
|
||||
addDirect(subscription);
|
||||
}
|
||||
if (LOG.isDebugEnabled()) {
|
||||
@@ -99,7 +99,7 @@ public class SubscriptionsStore {
|
||||
}
|
||||
|
||||
protected void addDirect(Subscription newSubscription) {
|
||||
TreeNode current = findMatchingNode(newSubscription.topic);
|
||||
TreeNode current = findMatchingNode(newSubscription.topicFilter);
|
||||
current.addSubscription(newSubscription);
|
||||
}
|
||||
|
||||
@@ -146,7 +146,7 @@ public class SubscriptionsStore {
|
||||
//search for the subscription to remove
|
||||
Subscription toBeRemoved = null;
|
||||
for (Subscription sub : matchNode.subscriptions()) {
|
||||
if (sub.topic.equals(topic) && sub.getClientId().equals(clientID)) {
|
||||
if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) {
|
||||
toBeRemoved = sub;
|
||||
break;
|
||||
}
|
||||
@@ -166,7 +166,7 @@ public class SubscriptionsStore {
|
||||
|
||||
List<Subscription> allSubscriptions = subsCollector.getResult();
|
||||
for (Subscription subscription : allSubscriptions) {
|
||||
removeSubscription(subscription.getTopic(), subscription.getClientId());
|
||||
removeSubscription(subscription.getTopicFilter(), subscription.getClientId());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -211,7 +211,7 @@ public class SubscriptionsStore {
|
||||
}
|
||||
|
||||
public boolean contains(Subscription sub) {
|
||||
return !matches(sub.topic).isEmpty();
|
||||
return !matches(sub.topicFilter).isEmpty();
|
||||
}
|
||||
|
||||
public int size() {
|
||||
|
||||
Reference in New Issue
Block a user