Browse Source

implemented removing all subscriptions by batch

pull/11693/head
YevhenBondarenko 2 years ago
parent
commit
8d04c5040e
  1. 39
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java
  2. 64
      application/src/main/java/org/thingsboard/server/service/subscription/TbEntityLocalSubsInfo.java

39
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java

@ -282,7 +282,6 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
if (sessionSubscriptions != null) {
TbSubscription<?> subscription = sessionSubscriptions.remove(subscriptionId);
if (subscription != null) {
if (sessionSubscriptions.isEmpty()) {
subscriptionsBySessionId.remove(sessionId);
}
@ -304,22 +303,26 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
@Override
public void cancelAllSessionSubscriptions(TenantId tenantId, String sessionId) {
log.debug("[{}][{}] Going to remove session subscriptions.", tenantId, sessionId);
List<SubscriptionModificationResult> results = new ArrayList<>();
Lock subsLock = getSubsLock(tenantId);
subsLock.lock();
try {
Map<Integer, TbSubscription<?>> sessionSubscriptions = subscriptionsBySessionId.remove(sessionId);
if (sessionSubscriptions != null) {
for (TbSubscription<?> subscription : sessionSubscriptions.values()) {
results.add(modifySubscription(tenantId, subscription.getEntityId(), subscription, false));
}
Map<EntityId, List<TbSubscription<?>>> entitySubscriptions =
sessionSubscriptions.values().stream().collect(Collectors.groupingBy(TbSubscription::getEntityId));
entitySubscriptions.forEach((entityId, subscriptions) -> {
TbEntitySubEvent event = removeAllSubscriptions(tenantId, entityId, subscriptions);
if (event != null) {
pushSubscriptionsEvent(tenantId, entityId, event);
}
});
} else {
log.debug("[{}][{}] No session subscriptions found!", tenantId, sessionId);
}
} finally {
subsLock.unlock();
}
results.stream().filter(SubscriptionModificationResult::hasEvent).forEach(this::pushSubscriptionEvent);
}
@Override
@ -500,6 +503,30 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
return new SubscriptionModificationResult(tenantId, entityId, subscription, missedUpdatesCandidate, event);
}
private TbEntitySubEvent removeAllSubscriptions(TenantId tenantId, EntityId entityId, List<TbSubscription<?>> subscriptions) {
TbEntitySubEvent event = null;
try {
TbEntityLocalSubsInfo entitySubs = subscriptionsByEntityId.get(entityId.getId());
event = entitySubs.removeAll(subscriptions);
if (entitySubs.isEmpty()) {
subscriptionsByEntityId.remove(entityId.getId());
entityUpdates.remove(entityId.getId());
}
} catch (Exception e) {
log.warn("[{}][{}] Failed to remove all subscriptions {} due to ", tenantId, entityId, subscriptions, e);
}
return event;
}
private void pushSubscriptionsEvent(TenantId tenantId, EntityId entityId, TbEntitySubEvent event) {
try {
log.trace("[{}][{}] Event: {}", tenantId, entityId, event);
pushSubEventToManagerService(tenantId, entityId, event);
} catch (Exception e) {
log.warn("[{}][{}] Failed to push subscription event {} due to ", tenantId, entityId, event, e);
}
}
private void pushSubscriptionEvent(SubscriptionModificationResult modificationResult) {
try {
TbEntitySubEvent event = modificationResult.getEvent();

64
application/src/main/java/org/thingsboard/server/service/subscription/TbEntityLocalSubsInfo.java

@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@ -129,13 +130,64 @@ public class TbEntityLocalSubsInfo {
if (!subs.remove(sub)) {
return null;
}
if (subs.isEmpty()) {
if (isEmpty()) {
return toEvent(ComponentLifecycleEvent.DELETED);
}
TbSubscriptionsInfo oldState = state.copy();
TbSubscriptionsInfo newState = new TbSubscriptionsInfo();
TbSubscriptionType type = sub.getType();
TbSubscriptionsInfo newState = state.copy();
updateNewState(newState, type);
return updateState(Set.of(type), newState);
}
public TbEntitySubEvent removeAll(List<TbSubscription<?>> subsToRemove) {
Set<TbSubscriptionType> changedTypes = new HashSet<>();
TbSubscriptionsInfo newState = state.copy();
for (TbSubscription<?> sub : subsToRemove) {
log.trace("[{}][{}][{}] Removing: {}", tenantId, entityId, sub.getSubscriptionId(), sub);
if (!subs.remove(sub)) {
continue;
}
if (isEmpty()) {
return toEvent(ComponentLifecycleEvent.DELETED);
}
TbSubscriptionType type = sub.getType();
if (changedTypes.contains(type)) {
continue;
}
updateNewState(newState, type);
changedTypes.add(type);
}
return updateState(changedTypes, newState);
}
private void updateNewState(TbSubscriptionsInfo state, TbSubscriptionType type) {
switch (type) {
case NOTIFICATIONS:
case NOTIFICATIONS_COUNT:
state.notifications = false;
break;
case ALARMS:
state.alarms = false;
break;
case ATTRIBUTES:
state.attrAllKeys = false;
state.attrKeys = null;
break;
case TIMESERIES:
state.tsAllKeys = false;
state.tsKeys = null;
}
}
private TbEntitySubEvent updateState(Set<TbSubscriptionType> updatedTypes, TbSubscriptionsInfo newState) {
for (TbSubscription<?> subscription : subs) {
switch (subscription.getType()) {
TbSubscriptionType type = subscription.getType();
if (!updatedTypes.contains(type)) {
continue;
}
switch (type) {
case NOTIFICATIONS:
case NOTIFICATIONS_COUNT:
if (!newState.notifications) {
@ -173,7 +225,7 @@ public class TbEntityLocalSubsInfo {
break;
}
}
if (newState.equals(oldState)) {
if (newState.equals(state)) {
return null;
} else {
this.state = newState;
@ -196,7 +248,7 @@ public class TbEntityLocalSubsInfo {
public boolean isEmpty() {
return state.isEmpty();
return subs.isEmpty();
}
public TbSubscription<?> registerPendingSubscription(TbSubscription<?> subscription, TbEntitySubEvent event) {

Loading…
Cancel
Save