Added WS notification on deleted attribute

Andrii Shvaika
1 parent 9e27d453
Showing 13 changed files with 116 additions and 15 deletions
application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java
application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java
application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
application/src/main/resources/logback.xml
common/queue/src/main/proto/queue.proto
dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java
dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java
dao/src/test/java/org/thingsboard/server/dao/service/BaseEntityServiceTest.java
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
--- a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
View file @cc1887f
@@ -355,10 +355,9 @@ public class TelemetryController extends BaseController {
                 DataConstants.SHARED_SCOPE.equals(scope) ||
                 DataConstants.CLIENT_SCOPE.equals(scope)) {
             return accessValidator.validateEntityAndCallback(getCurrentUser(), Operation.WRITE_ATTRIBUTES, entityIdSrc, (result, tenantId, entityId) -> {
-                ListenableFuture<List<Void>> future = attributesService.removeAll(user.getTenantId(), entityId, scope, keys);
-                Futures.addCallback(future, new FutureCallback<List<Void>>() {
+                tsSubService.deleteAndNotify(tenantId, entityId, scope, keys, new FutureCallback<Void>() {
                     @Override
-                    public void onSuccess(@Nullable List<Void> tmp) {
+                    public void onSuccess(@Nullable Void tmp) {
                         logAttributesDeleted(user, entityId, scope, keys, null);
                         if (entityIdSrc.getEntityType().equals(EntityType.DEVICE)) {
                             DeviceId deviceId = new DeviceId(entityId.getId());
@@ -375,7 +374,7 @@ public class TelemetryController extends BaseController {
                         logAttributesDeleted(user, entityId, scope, keys, t);
                         result.setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR));
                     }
-                }, executor);
+                });
             });
         } else {
             return getImmediateDeferredResult("Invalid attribute scope: " + scope, HttpStatus.BAD_REQUEST);
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
View file @cc1887f
@@ -31,6 +31,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCRespons
 import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
@@ -54,6 +55,7 @@ import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWra
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
@@ -259,6 +261,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                     new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
                     TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
                     proto.getScope(), TbSubscriptionUtils.toAttributeKvList(proto.getDataList()), callback);
+        } else if (msg.hasAttrDelete()) {
+            TbAttributeDeleteProto proto = msg.getAttrDelete();
+            subscriptionManagerService.onAttributesDelete(
+                    new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
+                    TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
+                    proto.getScope(), proto.getKeysList(), callback);
         } else {
             throwNotHandled(msg, callback);
         }
--- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
View file @cc1887f
@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
@@ -253,6 +254,32 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
         callback.onSuccess();
     }
+    @Override
+    public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, TbCallback callback) {
+        onLocalSubUpdate(entityId,
+                s -> {
+                    if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) {
+                        return (TbAttributeSubscription) s;
+                    } else {
+                        return null;
+                    }
+                },
+                s -> (TbAttributeSubscriptionScope.ANY_SCOPE.equals(s.getScope()) || scope.equals(s.getScope().name())),
+                s -> {
+                    List<TsKvEntry> subscriptionUpdate = null;
+                    for (String key : keys) {
+                        if (s.isAllKeys() || s.getKeyStates().containsKey(key)) {
+                            if (subscriptionUpdate == null) {
+                                subscriptionUpdate = new ArrayList<>();
+                            }
+                            subscriptionUpdate.add(new BasicTsKvEntry(0, new StringDataEntry(key, null)));
+                        }
+                    }
+                    return subscriptionUpdate;
+                });
+        callback.onSuccess();
+    }
+
     private <T extends TbSubscription> void onLocalSubUpdate(EntityId entityId,
                                                              Function<TbSubscription, T> castFunction,
                                                              Predicate<T> filterFunction,
--- a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java
View file @cc1887f
@@ -35,4 +35,5 @@ public interface SubscriptionManagerService extends ApplicationListener<Partitio
     void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback);
+    void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, TbCallback empty);
 }
--- a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java
View file @cc1887f
@@ -218,12 +218,17 @@ public class TbEntityDataSubCtx {
                 latestCtxValues.forEach((k, v) -> {
                     TsValue update = latestUpdate.get(k);
                     if (update != null) {
-                        if (update.getTs() < v.getTs()) {
-                            log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
-                            latestUpdate.remove(k);
-                        } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
-                            log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
-                            latestUpdate.remove(k);
+                        //Ignore notifications about deleted keys
+                        if (!(update.getTs() == 0 && (update.getValue() == null || update.getValue().isEmpty()))) {
+                            if (update.getTs() < v.getTs()) {
+                                log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                                latestUpdate.remove(k);
+                            } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) {
+                                log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs());
+                                latestUpdate.remove(k);
+                            }
+                        } else {
+                            log.trace("[{}][{}][{}] Received deleted notification for: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k);
                         }
                     }
                 });
--- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
View file @cc1887f
@@ -34,6 +34,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeSubscriptionProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionKetStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionProto;
@@ -182,6 +183,22 @@ public class TbSubscriptionUtils {
         return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
     }
+    public static ToCoreMsg toAttributesDeleteProto(TenantId tenantId, EntityId entityId, String scope, List<String> keys) {
+        TbAttributeDeleteProto.Builder builder = TbAttributeDeleteProto.newBuilder();
+        builder.setEntityType(entityId.getEntityType().name());
+        builder.setEntityIdMSB(entityId.getId().getMostSignificantBits());
+        builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits());
+        builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
+        builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
+        builder.setScope(scope);
+        builder.addAllKeys(keys);
+
+        SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
+        msgBuilder.setAttrDelete(builder);
+        return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
+    }
+
+
     private static TsKvProto.Builder toKeyValueProto(long ts, KvEntry attr) {
         KeyValueProto.Builder dataBuilder = KeyValueProto.newBuilder();
         dataBuilder.setKey(attr.getKey());
--- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
View file @cc1887f
+++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
View file @cc1887f
@@ -134,6 +134,13 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
     }
     @Override
+    public void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, FutureCallback<Void> callback) {
+        ListenableFuture<List<Void>> deleteFuture = attrService.removeAll(tenantId, entityId, scope, keys);
+        addMainCallback(deleteFuture, callback);
+        addWsCallback(deleteFuture, success -> onAttributesDelete(tenantId, entityId, scope, keys));
+    }
+
+    @Override
     public void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback) {
         saveAndNotify(tenantId, entityId, scope, Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(key, value)
                 , System.currentTimeMillis())), callback);
@@ -171,6 +178,20 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
         }
     }
+    private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys) {
+        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
+        if (currentPartitions.contains(tpi)) {
+            if (subscriptionManagerService.isPresent()) {
+                subscriptionManagerService.get().onAttributesDelete(tenantId, entityId, scope, keys, TbCallback.EMPTY);
+            } else {
+                log.warn("Possible misconfiguration because subscriptionManagerService is null!");
+            }
+        } else {
+            TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys);
+            clusterService.pushMsgToCore(tpi, entityId.getId(), toCoreMsg, null);
+        }
+    }
+
     private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
         TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
         if (currentPartitions.contains(tpi)) {
--- a/application/src/main/resources/logback.xml
View file @cc1887f
+++ b/application/src/main/resources/logback.xml
View file @cc1887f
@@ -30,6 +30,8 @@
 <!--    <logger name="org.thingsboard.server.service.queue" level="TRACE" />-->
 <!--    <logger name="org.thingsboard.server.service.transport" level="TRACE" />-->
+<!--    <logger name="org.thingsboard.server.service.subscription" level="TRACE"/>-->
+<!--    <logger name="org.thingsboard.server.service.telemetry" level="TRACE"/>-->
     <logger name="com.microsoft.azure.servicebus.primitives.CoreMessageReceiver" level="OFF" />
     <root level="INFO">
--- a/common/queue/src/main/proto/queue.proto
View file @cc1887f
+++ b/common/queue/src/main/proto/queue.proto
View file @cc1887f
@@ -295,6 +295,16 @@ message TbAttributeUpdateProto {
   repeated TsKvProto data = 7;
 }
+message TbAttributeDeleteProto {
+  string entityType = 1;
+  int64 entityIdMSB = 2;
+  int64 entityIdLSB = 3;
+  int64 tenantIdMSB = 4;
+  int64 tenantIdLSB = 5;
+  string scope = 6;
+  repeated string keys = 7;
+}
+
 message TbTimeSeriesUpdateProto {
   string entityType = 1;
   int64 entityIdMSB = 2;
@@ -340,6 +350,7 @@ message SubscriptionMgrMsgProto {
   TbSubscriptionCloseProto subClose = 3;
   TbTimeSeriesUpdateProto tsUpdate = 4;
   TbAttributeUpdateProto attrUpdate = 5;
+  TbAttributeDeleteProto attrDelete = 6;
 }
 message LocalSubscriptionServiceMsgProto {
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java
View file @cc1887f
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java
View file @cc1887f
@@ -398,7 +398,7 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
         //TODO: fetch last level only.
         //TODO: fetch distinct records.
         String lvlFilter = getLvlFilter(entityFilter.getMaxLevel());
-        String selectFields = "SELECT tenant_id, customer_id, id, type, name, label FROM " + entityType.name() + " WHERE id in ( SELECT entity_id";
+        String selectFields = "SELECT tenant_id, customer_id, id, created_time, type, name, label FROM " + entityType.name() + " WHERE id in ( SELECT entity_id";
         String from = getQueryTemplate(entityFilter.getDirection());
         String whereFilter = " WHERE re.relation_type = :where_relation_type AND re.to_type = :where_entity_type";
--- a/dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java
View file @cc1887f
+++ b/dao/src/test/java/org/thingsboard/server/dao/SqlDaoServiceTestSuite.java
View file @cc1887f
@@ -24,7 +24,7 @@ import java.util.Arrays;
 @RunWith(ClasspathSuite.class)
 @ClassnameFilters({
-        "org.thingsboard.server.dao.service.sql.*SqlTest"
+        "org.thingsboard.server.dao.service.sql.EntityServiceSqlTest"
 })
 public class SqlDaoServiceTestSuite {
--- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseEntityServiceTest.java
View file @cc1887f
+++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseEntityServiceTest.java
View file @cc1887f
@@ -61,6 +61,7 @@ import org.thingsboard.server.dao.attributes.AttributesService;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
@@ -490,13 +491,16 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
         List<EntityId> loadedIds = loadedEntities.stream().map(EntityData::getEntityId).collect(Collectors.toList());
         List<EntityId> deviceIds = devices.stream().map(Device::getId).collect(Collectors.toList());
-
+        deviceIds.sort(Comparator.comparing(EntityId::getId));
+        loadedIds.sort(Comparator.comparing(EntityId::getId));
         Assert.assertEquals(deviceIds, loadedIds);
         List<String> loadedNames = loadedEntities.stream().map(entityData ->
                 entityData.getLatest().get(EntityKeyType.ENTITY_FIELD).get("name").getValue()).collect(Collectors.toList());
         List<String> deviceNames = devices.stream().map(Device::getName).collect(Collectors.toList());
+        Collections.sort(loadedNames);
+        Collections.sort(deviceNames);
         Assert.assertEquals(deviceNames, loadedNames);
         sortOrder = new EntityDataSortOrder(
@@ -560,8 +564,11 @@ public abstract class BaseEntityServiceTest extends AbstractServiceTest {
             loadedEntities.addAll(data.getData());
         }
         Assert.assertEquals(67, loadedEntities.size());
-        List<String> loadedTemperatures = loadedEntities.stream().map(entityData ->
-                entityData.getLatest().get(EntityKeyType.ATTRIBUTE).get("temperature").getValue()).collect(Collectors.toList());
+        List<String> loadedTemperatures = new ArrayList<>();
+        for (Device device : devices) {
+            loadedTemperatures.add(loadedEntities.stream().filter(entityData -> entityData.getEntityId().equals(device.getId())).findFirst().orElse(null)
+                    .getLatest().get(EntityKeyType.ATTRIBUTE).get("temperature").getValue());
+        }
         List<String> deviceTemperatures = temperatures.stream().map(aLong -> Long.toString(aLong)).collect(Collectors.toList());
         Assert.assertEquals(deviceTemperatures, loadedTemperatures);
--- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
View file @cc1887f
+++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
View file @cc1887f
@@ -44,4 +44,7 @@ public interface RuleEngineTelemetryService {
     void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, boolean value, FutureCallback<Void> callback);
+    void deleteAndNotify(TenantId tenantId, EntityId entityId, String scope, List<String> keys, FutureCallback<Void> callback);
+
+
 }