Merge remote-tracking branch 'origin/feature/edge' into feature/edge

@@ -61,7 +61,7 @@ public class EdgeEventController extends BaseController {
             EdgeId edgeId = new EdgeId(toUUID(strEdgeId));
             EdgeId edgeId = new EdgeId(toUUID(strEdgeId));
             checkEdgeId(edgeId, Operation.READ);
             checkEdgeId(edgeId, Operation.READ);
             TimePageLink pageLink = createPageLink(limit, startTime, endTime, ascOrder, offset);
             TimePageLink pageLink = createPageLink(limit, startTime, endTime, ascOrder, offset);
-            return checkNotNull(edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink));
+            return checkNotNull(edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, false));
         } catch (Exception e) {
         } catch (Exception e) {
             throw handleException(e);
             throw handleException(e);
         }
         }

@@ -111,7 +111,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
 
 
     @Override
     @Override
     public TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink) {
     public TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink) {
-        return edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink);
+        return edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, true);
     }
     }
 
 
     @Override
     @Override

@@ -16,9 +16,7 @@
 package org.thingsboard.server.dao.edge;
 package org.thingsboard.server.dao.edge;
 
 
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
-import org.thingsboard.server.common.data.edge.EdgeEventType;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.TimePageData;
 import org.thingsboard.server.common.data.page.TimePageData;
@@ -28,6 +26,5 @@ public interface EdgeEventService {
 
 
     ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent);
     ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent);
 
 
-    TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink);
-
+    TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate);
 }
 }

@@ -155,6 +155,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
             @Override
             @Override
             public void onCompleted() {
             public void onCompleted() {
                 log.debug("[{}] The rpc session was closed!", edgeKey);
                 log.debug("[{}] The rpc session was closed!", edgeKey);
+                onError.accept(new EdgeConnectionException("[" + edgeKey + "] The rpc session was closed!"));
             }
             }
         };
         };
     }
     }

@@ -35,7 +35,7 @@ import java.util.List;
 public class BaseEdgeEventService implements EdgeEventService {
 public class BaseEdgeEventService implements EdgeEventService {
 
 
     @Autowired
     @Autowired
-    public EdgeEventDao edgeEventDao;
+    private EdgeEventDao edgeEventDao;
 
 
     @Override
     @Override
     public ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent) {
     public ListenableFuture<EdgeEvent> saveAsync(EdgeEvent edgeEvent) {
@@ -44,8 +44,8 @@ public class BaseEdgeEventService implements EdgeEventService {
     }
     }
 
 
     @Override
     @Override
-    public TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink) {
-        List<EdgeEvent> events = edgeEventDao.findEdgeEvents(tenantId.getId(), edgeId, pageLink);
+    public TimePageData<EdgeEvent> findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate) {
+        List<EdgeEvent> events = edgeEventDao.findEdgeEvents(tenantId.getId(), edgeId, pageLink, withTsUpdate);
         return new TimePageData<>(events, pageLink);
         return new TimePageData<>(events, pageLink);
     }
     }
 
 

@@ -53,7 +53,7 @@ public class CassandraEdgeEventDao extends CassandraAbstractSearchTimeDao<EdgeEv
     }
     }
 
 
     @Override
     @Override
-    public List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink) {
+    public List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate) {
         return null;
         return null;
     }
     }
 }
 }

@@ -46,6 +46,6 @@ public interface EdgeEventDao extends Dao<EdgeEvent> {
      * @param pageLink the pageLink
      * @param pageLink the pageLink
      * @return the event list
      * @return the event list
      */
      */
-    List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink);
+    List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate);
 
 
 }
 }

@@ -26,6 +26,7 @@ import org.springframework.data.jpa.domain.Specification;
 import org.springframework.data.repository.CrudRepository;
 import org.springframework.data.repository.CrudRepository;
 import org.springframework.stereotype.Component;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.common.data.UUIDConverter;
 import org.thingsboard.server.common.data.UUIDConverter;
+import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.id.EdgeEventId;
 import org.thingsboard.server.common.data.id.EdgeEventId;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EdgeId;
@@ -75,9 +76,9 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTimeDao<EdgeEventEntit
     }
     }
 
 
     @Override
     @Override
-    public List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink) {
+    public List<EdgeEvent> findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate) {
         Specification<EdgeEventEntity> timeSearchSpec = JpaAbstractSearchTimeDao.getTimeSearchPageSpec(pageLink, "id");
         Specification<EdgeEventEntity> timeSearchSpec = JpaAbstractSearchTimeDao.getTimeSearchPageSpec(pageLink, "id");
-        Specification<EdgeEventEntity> fieldsSpec = getEntityFieldsSpec(tenantId, edgeId);
+        Specification<EdgeEventEntity> fieldsSpec = getEntityFieldsSpec(tenantId, edgeId, withTsUpdate);
         Sort.Direction sortDirection = pageLink.isAscOrder() ? Sort.Direction.ASC : Sort.Direction.DESC;
         Sort.Direction sortDirection = pageLink.isAscOrder() ? Sort.Direction.ASC : Sort.Direction.DESC;
         Pageable pageable = PageRequest.of(0, pageLink.getLimit(), sortDirection, ID_PROPERTY);
         Pageable pageable = PageRequest.of(0, pageLink.getLimit(), sortDirection, ID_PROPERTY);
         return DaoUtil.convertDataList(edgeEventRepository.findAll(Specification.where(timeSearchSpec).and(fieldsSpec), pageable).getContent());
         return DaoUtil.convertDataList(edgeEventRepository.findAll(Specification.where(timeSearchSpec).and(fieldsSpec), pageable).getContent());
@@ -95,7 +96,7 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTimeDao<EdgeEventEntit
         return Optional.of(DaoUtil.getData(edgeEventRepository.save(entity)));
         return Optional.of(DaoUtil.getData(edgeEventRepository.save(entity)));
     }
     }
 
 
-    private Specification<EdgeEventEntity> getEntityFieldsSpec(UUID tenantId, EdgeId edgeId) {
+    private Specification<EdgeEventEntity> getEntityFieldsSpec(UUID tenantId, EdgeId edgeId, boolean withTsUpdate) {
         return (root, criteriaQuery, criteriaBuilder) -> {
         return (root, criteriaQuery, criteriaBuilder) -> {
             List<Predicate> predicates = new ArrayList<>();
             List<Predicate> predicates = new ArrayList<>();
             if (tenantId != null) {
             if (tenantId != null) {
@@ -106,6 +107,10 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTimeDao<EdgeEventEntit
                 Predicate entityIdPredicate = criteriaBuilder.equal(root.get("edgeId"), UUIDConverter.fromTimeUUID(edgeId.getId()));
                 Predicate entityIdPredicate = criteriaBuilder.equal(root.get("edgeId"), UUIDConverter.fromTimeUUID(edgeId.getId()));
                 predicates.add(entityIdPredicate);
                 predicates.add(entityIdPredicate);
             }
             }
+            if (!withTsUpdate) {
+                Predicate edgeEventActionPredicate = criteriaBuilder.notEqual(root.get("edgeEventAction"), ActionType.TIMESERIES_UPDATED.name());
+                predicates.add(edgeEventActionPredicate);
+            }
             return criteriaBuilder.and(predicates.toArray(new Predicate[]{}));
             return criteriaBuilder.and(predicates.toArray(new Predicate[]{}));
         };
         };
     }
     }

@@ -19,6 +19,7 @@ import com.datastax.driver.core.utils.UUIDs;
 import org.junit.Assert;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.Test;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.edge.EdgeEventType;
 import org.thingsboard.server.common.data.edge.EdgeEventType;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.DeviceId;
@@ -82,7 +83,7 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest {
         EdgeEvent savedEdgeEvent3 = saveEdgeEventWithProvidedTime(eventTime + 2, edgeId, deviceId, tenantId);
         EdgeEvent savedEdgeEvent3 = saveEdgeEventWithProvidedTime(eventTime + 2, edgeId, deviceId, tenantId);
         saveEdgeEventWithProvidedTime(timeAfterEndTime, edgeId, deviceId, tenantId);
         saveEdgeEventWithProvidedTime(timeAfterEndTime, edgeId, deviceId, tenantId);
 
 
-        TimePageData<EdgeEvent> edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, new TimePageLink(2, startTime, endTime, false));
+        TimePageData<EdgeEvent> edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, new TimePageLink(2, startTime, endTime, false), true);
 
 
         Assert.assertNotNull(edgeEvents.getData());
         Assert.assertNotNull(edgeEvents.getData());
         Assert.assertTrue(edgeEvents.getData().size() == 2);
         Assert.assertTrue(edgeEvents.getData().size() == 2);
@@ -91,7 +92,7 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest {
         Assert.assertTrue(edgeEvents.hasNext());
         Assert.assertTrue(edgeEvents.hasNext());
         Assert.assertNotNull(edgeEvents.getNextPageLink());
         Assert.assertNotNull(edgeEvents.getNextPageLink());
 
 
-        edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, edgeEvents.getNextPageLink());
+        edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, edgeEvents.getNextPageLink(), true);
 
 
         Assert.assertNotNull(edgeEvents.getData());
         Assert.assertNotNull(edgeEvents.getData());
         Assert.assertTrue(edgeEvents.getData().size() == 1);
         Assert.assertTrue(edgeEvents.getData().size() == 1);
@@ -100,6 +101,26 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest {
         Assert.assertNull(edgeEvents.getNextPageLink());
         Assert.assertNull(edgeEvents.getNextPageLink());
     }
     }
 
 
+    @Test
+    public void findEdgeEventsWithTsUpdateAndWithout() throws Exception {
+        EdgeId edgeId = new EdgeId(UUIDs.timeBased());
+        DeviceId deviceId = new DeviceId(UUIDs.timeBased());
+        TenantId tenantId = new TenantId(UUIDs.timeBased());
+        TimePageLink pageLink = new TimePageLink(1);
+
+        EdgeEvent edgeEventWithTsUpdate = generateEdgeEvent(tenantId, edgeId, deviceId, ActionType.TIMESERIES_UPDATED.name());
+        edgeEventService.saveAsync(edgeEventWithTsUpdate);
+
+        TimePageData<EdgeEvent> allEdgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, true);
+        TimePageData<EdgeEvent> edgeEventsWithoutTsUpdate = edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, false);
+
+        Assert.assertNotNull(allEdgeEvents.getData());
+        Assert.assertNotNull(edgeEventsWithoutTsUpdate.getData());
+        Assert.assertEquals(1, allEdgeEvents.getData().size());
+        Assert.assertEquals(allEdgeEvents.getData().get(0).getUuidId(), edgeEventWithTsUpdate.getUuidId());
+        Assert.assertTrue(edgeEventsWithoutTsUpdate.getData().isEmpty());
+    }
+
     private EdgeEvent saveEdgeEventWithProvidedTime(long time, EdgeId edgeId, EntityId entityId, TenantId tenantId) throws Exception {
     private EdgeEvent saveEdgeEventWithProvidedTime(long time, EdgeId edgeId, EntityId entityId, TenantId tenantId) throws Exception {
         EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, entityId, DataConstants.ENTITY_CREATED);
         EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, entityId, DataConstants.ENTITY_CREATED);
         edgeEvent.setId(new EdgeEventId(UUIDs.startOf(time)));
         edgeEvent.setId(new EdgeEventId(UUIDs.startOf(time)));