refactor: 设备事件上报逻辑调整

1、从设备配置中取物模型中的事件
2、设备事件表的事件标识、事件名称、事件类型从物模型获取
3、设备配置增加物模型

@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.*;
+import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
@@ -235,13 +236,13 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
 
     //Thingskit function
     @Override
-    public void saveAndNotify(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData, Long eventTime, FutureCallback<Void> callback) {
+    public void saveAndNotify(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, List<TkEventKvEntry> events, String eventData, Long eventTime, FutureCallback<Void> callback) {
         checkInternalEntity(entityId);
-        saveAndNotifyInternal(tenantId,profileId, entityId,  eventId,eventType, eventData,eventTime, callback);
+        saveAndNotifyInternal(tenantId,profileId, entityId,  events, eventData,eventTime, callback);
     }
     @Override
-    public void saveAndNotifyInternal(TenantId tenantId,DeviceProfileId profileId, EntityId entityId,String eventId, DeviceEventTypeEnum eventType, String eventData,Long eventTime, FutureCallback<Void> callback) {
-        ListenableFuture<List<Void>> saveFuture = eventsService.save(tenantId,profileId, entityId,eventId,eventType, eventData,eventTime);
+    public void saveAndNotifyInternal(TenantId tenantId,DeviceProfileId profileId, EntityId entityId,List<TkEventKvEntry> events, String eventData,Long eventTime, FutureCallback<Void> callback) {
+        ListenableFuture<List<Void>> saveFuture = eventsService.save(tenantId,profileId, entityId,events, eventData,eventTime);
         addVoidCallback(saveFuture, callback);
 //        addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice));
     }

@@ -39,7 +39,7 @@ public interface InternalTelemetryService extends RuleEngineTelemetryService {
     void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
 
     //Thingskit function
-    void saveAndNotifyInternal(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData,Long eventTime, FutureCallback<Void> callback);
+    void saveAndNotifyInternal(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, List<TkEventKvEntry> eventKvEntry, String eventData, Long eventTime, FutureCallback<Void> callback);
 
     void saveLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 

@@ -22,6 +22,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.TimePageLink;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvDto;
+import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 
 import java.util.List;
@@ -31,7 +32,7 @@ import java.util.List;
  */
 public interface TkEventsService {
     PageData<TkEventKvDto> findEvents(EntityId entityId,String eventIdentifier,  DeviceEventTypeEnum eventType, TimePageLink pageLink);
-    ListenableFuture<List<Void>> save(TenantId tenantId, DeviceProfileId profileId,EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData,Long eventTime);
+    ListenableFuture<List<Void>> save(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, List<TkEventKvEntry> eventKvEntry, String eventData, Long eventTime);
 
 
 

@@ -18,6 +18,7 @@ package org.thingsboard.server.common.data.device.profile;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
+import org.thingsboard.server.common.data.yunteng.dto.ThingsModelDTO;
 
 import javax.validation.Valid;
 import java.io.Serializable;
@@ -38,4 +39,9 @@ public class DeviceProfileData implements Serializable {
     @ApiModelProperty(position = 4, value = "JSON array of alarm rules configuration per device profile")
     private List<DeviceProfileAlarm> alarms;
 
+    //Thingskit function
+    @Valid
+    @ApiModelProperty(position = 1, value = "JSON object of has published things model configuration")
+    private List<ThingsModelDTO> thingsModel;
+
 }

-/**
- * Copyright © 2016-2022 The Thingsboard Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
 package org.thingsboard.server.common.data.yunteng.dto;
 
-import org.thingsboard.server.common.data.kv.KvEntry;
+import lombok.Data;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
+
+import java.io.Serializable;
 
 /**
  * @author Andrew Shvayka
  */
-public interface TkEventKvEntry extends KvEntry {
+@Data
+public class TkEventKvEntry implements Serializable {
+
 
-    long getEventTime();
+    private DeviceEventTypeEnum eventType;
+    private  String eventName;
+    private String eventIdentifier;
 
 }

@@ -25,10 +25,13 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.TimePageLink;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvDto;
+import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
+import org.thingsboard.server.dao.attributes.AttributeUtils;
 import org.thingsboard.server.dao.yunteng.event.TkEventsService;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.yunteng.jpa.dao.event.EventUtils.validate;
 
@@ -53,10 +56,10 @@ public class BaseEventsService implements TkEventsService {
     }
 
     @Override
-    public ListenableFuture<List<Void>> save(TenantId tenantId,DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType,  String eventData,Long eventTime) {
-        validate(entityId, eventId,eventType);
-
-        ListenableFuture<Void> saveFutures = eventsDao.save(tenantId, profileId,entityId, eventType,eventId, eventData,eventTime);
+    public ListenableFuture<List<Void>> save(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, List<TkEventKvEntry> events, String eventData, Long eventTime) {
+        validate(entityId);
+        events.forEach(event -> validate(event));
+        List<ListenableFuture<Void>> saveFutures = events.stream().map(event -> eventsDao.save(tenantId, profileId,entityId, event,eventData,eventTime)).collect(Collectors.toList());
         return Futures.allAsList(saveFutures);
     }
 

@@ -44,11 +44,8 @@ public class EventUtils {
     public static void validate(TkEventKvEntry kvEntry) {
         if (kvEntry == null) {
             throw new IncorrectParameterException("Key value entry can't be null");
-        } else if (kvEntry.getDataType() == null) {
-            throw new IncorrectParameterException("Incorrect kvEntry. Data type can't be null");
-        } else {
-            Validator.validateString(kvEntry.getKey(), "Incorrect kvEntry. Key can't be empty");
-            Validator.validatePositiveNumber(kvEntry.getEventTime(), "Incorrect last update ts. Ts should be positive");
+        } else if (kvEntry.getEventType() == null) {
+            throw new IncorrectParameterException("Incorrect event type. event type can't be null");
         }
     }
 }

@@ -22,8 +22,11 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.TimePageLink;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvDto;
+import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 
+import java.util.List;
+
 /**
  * @author Andrew Shvayka
  */
@@ -31,6 +34,6 @@ public interface EventsDao {
 
     PageData<TkEventKvDto> findEvents(EntityId entityId,String eventIdentifier,  DeviceEventTypeEnum eventType, TimePageLink pageLink);
 
-    ListenableFuture<Void> save(TenantId tenantId, DeviceProfileId profileId,EntityId entityId, DeviceEventTypeEnum attributeType,String eventIdentifier, String eventData,Long eventTime);
+    ListenableFuture<Void> save(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, TkEventKvEntry eventKvEntry, String eventData, Long eventTime);
 
 }

@@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.TimePageLink;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvDto;
+import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 import org.thingsboard.server.common.stats.StatsFactory;
 import org.thingsboard.server.dao.DaoUtil;
@@ -40,6 +41,7 @@ import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvEntity;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.Comparator;
+import java.util.List;
 import java.util.function.Function;
 
 @Component
@@ -111,10 +113,10 @@ public class JpaEventDao extends JpaAbstractDaoListeningExecutorService implemen
 
 
     @Override
-    public ListenableFuture<Void> save(TenantId tenantId, DeviceProfileId profileId,EntityId entityId,DeviceEventTypeEnum eventType, String eventIdentifier, String eventData,Long eventTime) {
+    public ListenableFuture<Void> save(TenantId tenantId, DeviceProfileId profileId, EntityId entityId,TkEventKvEntry eventKvEntry,  String eventData, Long eventTime) {
         TkEventKvEntity entity = new TkEventKvEntity();
-        entity.setId(new TkEventKvCompositeKey(eventType, entityId.getId(), eventIdentifier, eventTime));
-        entity.setEventName("test");
+        entity.setId(new TkEventKvCompositeKey(eventKvEntry.getEventType(), entityId.getId(), eventKvEntry.getEventIdentifier(), eventTime));
+        entity.setEventName(eventKvEntry.getEventName());
         entity.setDeviceProfileId(profileId.getId());
         entity.setEventValue(eventData);
         return addToQueue(entity);

@@ -45,7 +45,7 @@ public interface RuleEngineTelemetryService {
     void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);
 
     //Thingskit function
-    void saveAndNotify(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData,Long eventTime, FutureCallback<Void> callback);
+    void saveAndNotify(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, List<TkEventKvEntry> eventKvEntry, String eventData, Long eventTime, FutureCallback<Void> callback);
 
     void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 

@@ -19,13 +19,21 @@ import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.rule.engine.api.*;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.rule.engine.telemetry.TelemetryNodeCallback;
+import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
+import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.data.yunteng.dto.ThingsModelDTO;
+import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
+import org.thingsboard.server.common.data.yunteng.enums.FunctionTypeEnum;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.session.SessionMsgType;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.model.ModelConstants.*;
 
@@ -57,17 +65,29 @@ public class TkMsgEventNode implements TbNode {
             ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType()));
             return;
         }
-        String eventIdentifier = msg.getMetaData().getValue(EVENT_IDENTIFIER_COLUMN);
         String deviceProfileId = msg.getMetaData().getValue(DEVICE_DEVICE_PROFILE_ID_PROPERTY);
+        TenantId tenantId = ctx.getTenantId();
+        DeviceProfileId profileId = new DeviceProfileId(UUID.fromString(deviceProfileId));
+        DeviceProfile profile = cache.get(tenantId,profileId);
+        String eventIdentifier = msg.getMetaData().getValue(EVENT_IDENTIFIER_COLUMN);
+        List<ThingsModelDTO> eventList =profile.getProfileData().getThingsModel().stream()
+                .filter(f -> f.getIdentifier().equals(eventIdentifier) && FunctionTypeEnum.events.equals(f.getFunctionType()))
+                .collect(Collectors.toList());
+        List<TkEventKvEntry> entryList = new ArrayList<>();
+        eventList.stream().forEach(i ->{
+            TkEventKvEntry item = new TkEventKvEntry();
+            item.setEventIdentifier(i.getIdentifier());
+            item.setEventType(i.getEventType());
+            item.setEventName(i.getFunctionName());
+        });
         ////TODO: 验证事件类型、事件标识符和数据建是否与产品物模型中的事件匹配
         long ts = System.currentTimeMillis();
         String src = msg.getData();
         DeviceEventTypeEnum eventType = DeviceEventTypeEnum.valueOf(msg.getMetaData().getValue(EVENT_TYPE_COLUMN));
         ctx.getTelemetryService().saveAndNotify(
-                ctx.getTenantId(),new DeviceProfileId(UUID.fromString(deviceProfileId)),
+                tenantId,profileId,
                 msg.getOriginator(),
-                eventIdentifier,
-                eventType,
+                entryList,
                 src,ts,
                 new TelemetryNodeCallback(ctx, msg)
         );