Showing
1 changed file
with
9 additions
and
8 deletions
@@ -52,6 +52,7 @@ import java.util.Map; | @@ -52,6 +52,7 @@ import java.util.Map; | ||
52 | import java.util.Optional; | 52 | import java.util.Optional; |
53 | import java.util.Set; | 53 | import java.util.Set; |
54 | import java.util.concurrent.ConcurrentHashMap; | 54 | import java.util.concurrent.ConcurrentHashMap; |
55 | +import java.util.concurrent.ConcurrentMap; | ||
55 | import java.util.concurrent.TimeUnit; | 56 | import java.util.concurrent.TimeUnit; |
56 | import java.util.concurrent.atomic.AtomicInteger; | 57 | import java.util.concurrent.atomic.AtomicInteger; |
57 | 58 | ||
@@ -62,11 +63,11 @@ import java.util.concurrent.atomic.AtomicInteger; | @@ -62,11 +63,11 @@ import java.util.concurrent.atomic.AtomicInteger; | ||
62 | final class MqttClientImpl implements MqttClient { | 63 | final class MqttClientImpl implements MqttClient { |
63 | 64 | ||
64 | private final Set<String> serverSubscriptions = new HashSet<>(); | 65 | private final Set<String> serverSubscriptions = new HashSet<>(); |
65 | - private final ConcurrentHashMap<Integer, MqttPendingUnsubscription> pendingServerUnsubscribes = new ConcurrentHashMap<>(); | ||
66 | - private final ConcurrentHashMap<Integer, MqttIncomingQos2Publish> qos2PendingIncomingPublishes = new ConcurrentHashMap<>(); | ||
67 | - private final ConcurrentHashMap<Integer, MqttPendingPublish> pendingPublishes = new ConcurrentHashMap<>(); | 66 | + private final ConcurrentMap<Integer, MqttPendingUnsubscription> pendingServerUnsubscribes = new ConcurrentHashMap<>(); |
67 | + private final ConcurrentMap<Integer, MqttIncomingQos2Publish> qos2PendingIncomingPublishes = new ConcurrentHashMap<>(); | ||
68 | + private final ConcurrentMap<Integer, MqttPendingPublish> pendingPublishes = new ConcurrentHashMap<>(); | ||
68 | private final HashMultimap<String, MqttSubscription> subscriptions = HashMultimap.create(); | 69 | private final HashMultimap<String, MqttSubscription> subscriptions = HashMultimap.create(); |
69 | - private final ConcurrentHashMap<Integer, MqttPendingSubscription> pendingSubscriptions = new ConcurrentHashMap<>(); | 70 | + private final ConcurrentMap<Integer, MqttPendingSubscription> pendingSubscriptions = new ConcurrentHashMap<>(); |
70 | private final Set<String> pendingSubscribeTopics = new HashSet<>(); | 71 | private final Set<String> pendingSubscribeTopics = new HashSet<>(); |
71 | private final HashMultimap<MqttHandler, MqttSubscription> handlerToSubscribtion = HashMultimap.create(); | 72 | private final HashMultimap<MqttHandler, MqttSubscription> handlerToSubscribtion = HashMultimap.create(); |
72 | private final AtomicInteger nextMessageId = new AtomicInteger(1); | 73 | private final AtomicInteger nextMessageId = new AtomicInteger(1); |
@@ -489,7 +490,7 @@ final class MqttClientImpl implements MqttClient { | @@ -489,7 +490,7 @@ final class MqttClientImpl implements MqttClient { | ||
489 | } | 490 | } |
490 | } | 491 | } |
491 | 492 | ||
492 | - ConcurrentHashMap<Integer, MqttPendingSubscription> getPendingSubscriptions() { | 493 | + ConcurrentMap<Integer, MqttPendingSubscription> getPendingSubscriptions() { |
493 | return pendingSubscriptions; | 494 | return pendingSubscriptions; |
494 | } | 495 | } |
495 | 496 | ||
@@ -509,15 +510,15 @@ final class MqttClientImpl implements MqttClient { | @@ -509,15 +510,15 @@ final class MqttClientImpl implements MqttClient { | ||
509 | return serverSubscriptions; | 510 | return serverSubscriptions; |
510 | } | 511 | } |
511 | 512 | ||
512 | - ConcurrentHashMap<Integer, MqttPendingUnsubscription> getPendingServerUnsubscribes() { | 513 | + ConcurrentMap<Integer, MqttPendingUnsubscription> getPendingServerUnsubscribes() { |
513 | return pendingServerUnsubscribes; | 514 | return pendingServerUnsubscribes; |
514 | } | 515 | } |
515 | 516 | ||
516 | - ConcurrentHashMap<Integer, MqttPendingPublish> getPendingPublishes() { | 517 | + ConcurrentMap<Integer, MqttPendingPublish> getPendingPublishes() { |
517 | return pendingPublishes; | 518 | return pendingPublishes; |
518 | } | 519 | } |
519 | 520 | ||
520 | - ConcurrentHashMap<Integer, MqttIncomingQos2Publish> getQos2PendingIncomingPublishes() { | 521 | + ConcurrentMap<Integer, MqttIncomingQos2Publish> getQos2PendingIncomingPublishes() { |
521 | return qos2PendingIncomingPublishes; | 522 | return qos2PendingIncomingPublishes; |
522 | } | 523 | } |
523 | 524 |