Apache/activemq-artemis
1
0
Fork 0
mirror of https://github.com/apache/activemq-artemis.git synced 2025-03-08 18:29:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

This closes #1294

Browse Source
This commit is contained in:
Martyn Taylor 2017-05-26 11:16:42 +01:00
commit de9cc2e708
1 changed files with 14 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java
View File

@ -41,7 +41,7 @@ public class MQTTSessionState {
private final ConcurrentMap<String, Map<Long, Integer>> addressMessageMap = new ConcurrentHashMap<>();
private final Set<Integer> pubRec = new HashSet<>();
private final Set<Integer> pubRec = new HashSet<>();
private boolean attached = false;
@ -128,7 +128,7 @@ public class MQTTSessionState {
public class OutboundStore {
private HashMap<String, Integer> artemisToMqttMessageMap = new HashMap<>();
private HashMap<Pair<Long, Long>, Integer> artemisToMqttMessageMap = new HashMap<>();
private HashMap<Integer, Pair<Long, Long>> mqttToServerIds = new HashMap<>();
@ -136,9 +136,13 @@ public class MQTTSessionState {
private final AtomicInteger ids = new AtomicInteger(0);
public int generateMqttId(long serverId, long consumerId) {
private Pair<Long, Long> generateKey(long messageId, long consumerID) {
return new Pair<>(messageId, consumerID);
}
public int generateMqttId(long messageId, long consumerId) {
synchronized (dataStoreLock) {
Integer id = artemisToMqttMessageMap.get(consumerId + ":" + serverId);
Integer id = artemisToMqttMessageMap.get(generateKey(messageId, consumerId));
if (id == null) {
ids.compareAndSet(Short.MAX_VALUE, 1);
id = ids.addAndGet(1);
@ -147,18 +151,20 @@ public class MQTTSessionState {
}
}
public void publish(int mqtt, long serverId, long consumerId) {
public void publish(int mqtt, long messageId, long consumerId) {
synchronized (dataStoreLock) {
artemisToMqttMessageMap.put(consumerId + ":" + serverId, mqtt);
mqttToServerIds.put(mqtt, new Pair(serverId, consumerId));
Pair<Long, Long> key = generateKey(messageId, consumerId);
artemisToMqttMessageMap.put(key, mqtt);
mqttToServerIds.put(mqtt, key);
}
}
public Pair<Long, Long> publishAckd(int mqtt) {
synchronized (dataStoreLock) {
Pair p = mqttToServerIds.remove(mqtt);
Pair p = mqttToServerIds.remove(mqtt);
if (p != null) {
mqttToServerIds.remove(p.getA());
artemisToMqttMessageMap.remove(p);
}
return p;
}