Merge branch '20230222' into 'master'

feat: 租户的规则链更新

See merge request yunteng/thingskit!159

@@ -113,6 +113,19 @@
           "port": "8082",
           "resetApi": "/api/v1/notice/alert"
         }
+      },
+      {
+        "additionalInfo": {
+          "description": "",
+          "layoutX": 824,
+          "layoutY": 549
+        },
+        "type": "org.thingsboard.rule.engine.yunteng.event.TkMsgEventNode",
+        "name": "save events",
+        "debugMode": false,
+        "configuration": {
+          "useServerTs": true
+        }
       }
     ],
     "connections": [
@@ -150,6 +163,11 @@
         "fromIndex": 2,
         "toIndex": 5,
         "type": "RPC Request to Device"
+      },
+      {
+        "fromIndex": 2,
+        "toIndex": 8,
+        "type": "Post event"
       }
     ],
     "ruleChainConnections": null

@@ -316,7 +316,7 @@ public abstract class BaseController {
         }
     }
 
-    void checkParameter(String name, String param) throws ThingsboardException {
+    protected void checkParameter(String name, String param) throws ThingsboardException {
         if (StringUtils.isEmpty(param)) {
             throw new ThingsboardException("参数【 " + name + "】不能为空!", ThingsboardErrorCode.BAD_REQUEST_PARAMS);
         }

+package org.thingsboard.server.controller.yunteng;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.common.data.id.DeviceId;
+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.enums.DeviceEventTypeEnum;
+import org.thingsboard.server.common.data.yunteng.enums.OrderTypeEnum;
+import org.thingsboard.server.controller.BaseController;
+import org.thingsboard.server.dao.model.ModelConstants;
+import org.thingsboard.server.dao.yunteng.event.TkEventsService;
+
+import java.util.UUID;
+
+import static org.thingsboard.server.common.data.yunteng.constant.QueryConstant.*;
+
+/**
+ * @author Administrator
+ */
+@RestController
+@RequestMapping("api/yt/event")
+@Api(tags = {"设备事件"})
+@RequiredArgsConstructor
+public class TkEventController extends BaseController {
+    private final TkEventsService eventsService;
+
+    @GetMapping(params = {PAGE_SIZE, PAGE})
+    @ApiOperation("分页")
+    public PageData<TkEventKvDto> pageAlarmProfile(
+            @RequestParam(PAGE_SIZE) int pageSize,
+            @RequestParam(PAGE) int page,
+            @RequestParam(value = "eventIdentifier", required = false) String eventIdentifier,
+            @RequestParam(value = "eventType", required = false) DeviceEventTypeEnum eventType,
+            @RequestParam(value = "tbDeviceId", required = true) String tbDeviceId,
+            @RequestParam(required = false) Long startTime,
+            @RequestParam(required = false) Long endTime,
+            @RequestParam(value = ORDER_FILED, required = false) String orderBy,
+            @RequestParam(value = ORDER_TYPE, required = false) OrderTypeEnum orderType)
+            throws ThingsboardException {
+        checkParameter("tbDeviceId", tbDeviceId);
+
+        if (orderBy != null && orderBy.isEmpty()) {
+            orderBy = ModelConstants.EVENT_TIME_COLUMN;
+        }
+        TimePageLink pageLink = createTimePageLink(pageSize, page, null, orderBy, orderType==null?"":orderType.name(), startTime, endTime);
+        return checkNotNull(eventsService.findEvents(new DeviceId(UUID.fromString(tbDeviceId)),eventIdentifier,eventType, pageLink));
+    }
+
+
+}

@@ -28,19 +28,11 @@ import org.thingsboard.server.common.data.ApiUsageRecordKey;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EntityView;
 import org.thingsboard.server.common.data.id.CustomerId;
+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.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
-import org.thingsboard.server.common.data.kv.BooleanDataEntry;
-import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
-import org.thingsboard.server.common.data.kv.DoubleDataEntry;
-import org.thingsboard.server.common.data.kv.LongDataEntry;
-import org.thingsboard.server.common.data.kv.StringDataEntry;
-import org.thingsboard.server.common.data.kv.TsKvEntry;
-import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult;
-import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+import org.thingsboard.server.common.data.kv.*;
+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;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@@ -57,15 +49,7 @@ import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
 import javax.annotation.Nullable;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -251,13 +235,13 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
 
     //Thingskit function
     @Override
-    public void saveAndNotify(TenantId tenantId, EntityId entityId, String eventId, TkEventType eventType, List<TkEventKvEntry> attributes, FutureCallback<Void> callback) {
+    public void saveAndNotify(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData, Long eventTime, FutureCallback<Void> callback) {
         checkInternalEntity(entityId);
-        saveAndNotifyInternal(tenantId, entityId,  eventId,eventType, attributes, callback);
+        saveAndNotifyInternal(tenantId,profileId, entityId,  eventId,eventType, eventData,eventTime, callback);
     }
     @Override
-    public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId,String eventId, TkEventType eventType, List<TkEventKvEntry> attributes, FutureCallback<Void> callback) {
-        ListenableFuture<List<Void>> saveFuture = eventsService.save(tenantId, entityId,eventId,eventType, attributes);
+    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);
         addVoidCallback(saveFuture, callback);
 //        addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice));
     }

@@ -17,12 +17,13 @@ package org.thingsboard.server.service.telemetry;
 
 import com.google.common.util.concurrent.FutureCallback;
 import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
+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.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 
 import java.util.List;
 
@@ -38,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, EntityId entityId, String eventId, TkEventType eventType, List<TkEventKvEntry> attributes, FutureCallback<Void> callback);
+    void saveAndNotifyInternal(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData,Long eventTime, FutureCallback<Void> callback);
 
     void saveLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 

@@ -16,34 +16,23 @@
 package org.thingsboard.server.dao.yunteng.event;
 
 import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.common.data.EntityType;
 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.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+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.enums.DeviceEventTypeEnum;
 
-import java.util.Collection;
 import java.util.List;
-import java.util.Optional;
 
 /**
  * @author Andrew Shvayka
  */
 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<Optional<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId,   String scope,String eventKey);
 
-    ListenableFuture<List<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId,  String scope, Collection<String> eventKeys);
-
-    ListenableFuture<List<TkEventKvEntry>> findAll(TenantId tenantId, EntityId entityId,String eventId,  TkEventType eventType);
-
-    ListenableFuture<List<Void>> save(TenantId tenantId, EntityId entityId, String eventId, TkEventType eventType, List<TkEventKvEntry> attributes);
-
-    ListenableFuture<List<Void>> removeAll(TenantId tenantId, EntityId entityId, String eventId, TkEventType eventType, List<String> eventKeys);
-
-    List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId);
-
-    List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds);
 
 }

@@ -15,107 +15,28 @@
  */
 package org.thingsboard.server.common.data.yunteng.dto;
 
-import org.thingsboard.server.common.data.kv.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.DataType;
-import org.thingsboard.server.common.data.kv.KvEntry;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.swagger.annotations.ApiModel;
+import lombok.Data;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 
-import javax.validation.Valid;
-import java.util.Optional;
+import java.util.UUID;
 
 /**
  * @author Andrew Shvayka
  */
-public class TkEventKvDto implements TkEventKvEntry {
+@Data
+@ApiModel
+public class TkEventKvDto {
 
     private static final long serialVersionUID = -6460767583563159407L;
 
     private final long eventTime;
-    @Valid
-    private final KvEntry kv;
-
-    public TkEventKvDto(KvEntry kv, long eventTime) {
-        this.kv = kv;
-        this.eventTime = eventTime;
-    }
-
-    public TkEventKvDto(long lastUpdateTs, KvEntry kv) {
-        this(kv, lastUpdateTs);
-    }
-
-    @Override
-    public long getEventTime() {
-        return eventTime;
-    }
-
-    @Override
-    public String getKey() {
-        return kv.getKey();
-    }
-
-    @Override
-    public DataType getDataType() {
-        return kv.getDataType();
-    }
-
-    @Override
-    public Optional<String> getStrValue() {
-        return kv.getStrValue();
-    }
-
-    @Override
-    public Optional<Long> getLongValue() {
-        return kv.getLongValue();
-    }
-
-    @Override
-    public Optional<Boolean> getBooleanValue() {
-        return kv.getBooleanValue();
-    }
-
-    @Override
-    public Optional<Double> getDoubleValue() {
-        return kv.getDoubleValue();
-    }
-
-    @Override
-    public Optional<String> getJsonValue() {
-        return kv.getJsonValue();
-    }
-
-    @Override
-    public String getValueAsString() {
-        return kv.getValueAsString();
-    }
-
-    @Override
-    public Object getValue() {
-        return kv.getValue();
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-
-        TkEventKvDto that = (TkEventKvDto) o;
-
-        if (eventTime != that.eventTime) return false;
-        return kv.equals(that.kv);
-
-    }
-
-    @Override
-    public int hashCode() {
-        int result = (int) (eventTime ^ (eventTime >>> 32));
-        result = 31 * result + kv.hashCode();
-        return result;
-    }
-
-    @Override
-    public String toString() {
-        return "BaseAttributeKvEntry{" +
-                "lastUpdateTs=" + eventTime +
-                ", kv=" + kv +
-                '}';
-    }
+    private final DeviceEventTypeEnum eventType;
+    private final JsonNode eventValue;
+    private final UUID entityId;
+    private final String eventIdentifier;
+    private final String eventName;
+    private final UUID deviceProfileId;
+    private final String deviceProfileName;
 }

-package org.thingsboard.server.common.data.yunteng.enums;
-
-public enum TkEventType {
-    INFO,ALARM,ERROR
-}

@@ -15,53 +15,18 @@
  */
 package org.thingsboard.server.common.transport.adaptor;
 
-import com.google.gson.Gson;
-import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
-import com.google.gson.JsonPrimitive;
-import com.google.gson.JsonSyntaxException;
+import com.google.gson.*;
 import org.apache.commons.lang3.math.NumberUtils;
 import org.springframework.util.StringUtils;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.kv.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
-import org.thingsboard.server.common.data.kv.BooleanDataEntry;
-import org.thingsboard.server.common.data.kv.DoubleDataEntry;
-import org.thingsboard.server.common.data.kv.JsonDataEntry;
-import org.thingsboard.server.common.data.kv.KvEntry;
-import org.thingsboard.server.common.data.kv.LongDataEntry;
-import org.thingsboard.server.common.data.kv.StringDataEntry;
-import org.thingsboard.server.common.data.yunteng.dto.TkEventKvDto;
-import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
+import org.thingsboard.server.common.data.kv.*;
 import org.thingsboard.server.gen.transport.TransportProtos;
-import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.CredentialsType;
-import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
-import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
-import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceResponseMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ResponseStatus;
-import org.thingsboard.server.gen.transport.TransportProtos.TsKvListProto;
-import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
-import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCredRequestMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.*;
 
 import java.math.BigDecimal;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
-import java.util.TreeMap;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -518,12 +483,6 @@ public class JsonConverter {
         return result;
     }
 
-    public static Set<TkEventKvEntry> convertToEvents(JsonElement element) {
-        Set<TkEventKvEntry> result = new HashSet<>();
-        long ts = System.currentTimeMillis();
-        result.addAll(parseValues(element.getAsJsonObject()).stream().map(kv -> new TkEventKvDto(kv, ts)).collect(Collectors.toList()));
-        return result;
-    }
 
     private static List<KvEntry> parseValues(JsonObject valuesObject) {
         List<KvEntry> result = new ArrayList<>();

@@ -593,6 +593,8 @@ public class DefaultTransportService implements TransportService {
         DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()));
         metaData.putValue("deviceName", sessionInfo.getDeviceName());
         metaData.putValue("deviceType", sessionInfo.getDeviceType());
+        UUID deviceProfileId = new UUID(sessionInfo.getDeviceProfileIdMSB(),sessionInfo.getDeviceProfileIdMSB());
+        metaData.putValue("device_profile_id", deviceProfileId.toString());
         metaData.putValue("deviceId",eventInfo[0]);
         metaData.putValue("event_identifier", eventInfo[1]);
         metaData.putValue("event_type",eventInfo[2]);

@@ -60,7 +60,7 @@ public class ModelConstants {
     public static final String EVENT_IDENTIFIER_COLUMN = "event_identifier";
     public static final String EVENT_TIME_COLUMN = "event_time";
     public static final String EVENT_TYPE_COLUMN = "event_type";
-    public static final String EVENT_KEY_COLUMN = "event_key";
+    public static final String EVENT_VALUE_COLUMN = "event_value";
 
     /**
      * Cassandra user constants.

@@ -19,19 +19,16 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
-import org.thingsboard.server.common.data.EntityType;
 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.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
-import org.thingsboard.server.dao.service.Validator;
+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.enums.DeviceEventTypeEnum;
 import org.thingsboard.server.dao.yunteng.event.TkEventsService;
 
-import java.util.Collection;
 import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.yunteng.jpa.dao.event.EventUtils.validate;
 
@@ -48,48 +45,19 @@ public class BaseEventsService implements TkEventsService {
         this.eventsDao = eventsDao;
     }
 
-    @Override
-    public ListenableFuture<Optional<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId,   String eventId,String attributeKey) {
-        validate(entityId, eventId);
-        Validator.validateString(attributeKey, "Incorrect attribute key " + attributeKey);
-        return eventsDao.find(tenantId, entityId, null, attributeKey);
-    }
 
     @Override
-    public ListenableFuture<List<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId, String eventId, Collection<String> attributeKeys) {
-        validate(entityId, eventId);
-        attributeKeys.forEach(attributeKey -> Validator.validateString(attributeKey, "Incorrect attribute key " + attributeKey));
-        return eventsDao.find(tenantId, entityId, null, attributeKeys);
+    public PageData<TkEventKvDto> findEvents(EntityId entityId, String eventId, DeviceEventTypeEnum eventType, TimePageLink pageLink) {
+        validate(entityId);
+        return eventsDao.findEvents( entityId, eventId,eventType,pageLink);
     }
 
     @Override
-    public ListenableFuture<List<TkEventKvEntry>> findAll(TenantId tenantId, EntityId entityId,String eventId, TkEventType eventType) {
-        validate(entityId,eventId);
-        return eventsDao.findAll(tenantId, entityId, eventType);
-    }
-
-    @Override
-    public List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId) {
-        return eventsDao.findAllKeysByDeviceProfileId(tenantId, deviceProfileId);
-    }
-
-    @Override
-    public List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds) {
-        return eventsDao.findAllKeysByEntityIds(tenantId, entityType, entityIds);
-    }
-
-    @Override
-    public ListenableFuture<List<Void>> save(TenantId tenantId, EntityId entityId, String eventId, TkEventType eventType,  List<TkEventKvEntry> attributes) {
+    public ListenableFuture<List<Void>> save(TenantId tenantId,DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType,  String eventData,Long eventTime) {
         validate(entityId, eventId,eventType);
-        attributes.forEach(attribute -> validate(attribute));
 
-        List<ListenableFuture<Void>> saveFutures = attributes.stream().map(attribute -> eventsDao.save(tenantId, entityId, eventType,eventId, attribute)).collect(Collectors.toList());
+        ListenableFuture<Void> saveFutures = eventsDao.save(tenantId, profileId,entityId, eventType,eventId, eventData,eventTime);
         return Futures.allAsList(saveFutures);
     }
 
-    @Override
-    public ListenableFuture<List<Void>> removeAll(TenantId tenantId, EntityId entityId, String eventId, TkEventType eventType, List<String> eventKeys) {
-        validate(entityId, eventId);
-        return eventsDao.removeAll(tenantId, entityId, eventId,eventType, eventKeys);
-    }
 }

@@ -16,20 +16,24 @@
 package org.thingsboard.server.dao.yunteng.jpa.dao.event;
 
 import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 import org.thingsboard.server.dao.exception.IncorrectParameterException;
 import org.thingsboard.server.dao.service.Validator;
 
 public class EventUtils {
+    public static void validate(EntityId id) {
+        Validator.validateId(id.getId(), "Incorrect id " + id);
+    }
+
+
     public static void validate(EntityId id, String eventIdentifier) {
         Validator.validateId(id.getId(), "Incorrect id " + id);
-        Validator.validateString(eventIdentifier, "Incorrect scope " + eventIdentifier);
+        Validator.validateString(eventIdentifier, "Incorrect event identifier " + eventIdentifier);
     }
 
 
-    public static void validate(EntityId id, String eventIdentifier, TkEventType eventType) {
+    public static void validate(EntityId id, String eventIdentifier, DeviceEventTypeEnum eventType) {
         Validator.validateId(id.getId(), "Incorrect id " + id);
         Validator.validateString(eventIdentifier, "Incorrect scope " + eventIdentifier);
         if(eventType == null){

@@ -16,33 +16,21 @@
 package org.thingsboard.server.dao.yunteng.jpa.dao.event;
 
 import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.common.data.EntityType;
 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.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Optional;
+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.enums.DeviceEventTypeEnum;
 
 /**
  * @author Andrew Shvayka
  */
 public interface EventsDao {
 
-    ListenableFuture<Optional<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId, TkEventType attributeType, String attributeKey);
-
-    ListenableFuture<List<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId, TkEventType attributeType, Collection<String> attributeKey);
-
-    ListenableFuture<List<TkEventKvEntry>> findAll(TenantId tenantId, EntityId entityId, TkEventType attributeType);
-
-    ListenableFuture<Void> save(TenantId tenantId, EntityId entityId, TkEventType attributeType,String eventIdentifier, TkEventKvEntry attribute);
-
-    ListenableFuture<List<Void>> removeAll(TenantId tenantId, EntityId entityId, String eventId,TkEventType eventType,List<String> keys);
+    PageData<TkEventKvDto> findEvents(EntityId entityId,String eventIdentifier,  DeviceEventTypeEnum eventType, TimePageLink pageLink);
 
-    List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId);
+    ListenableFuture<Void> save(TenantId tenantId, DeviceProfileId profileId,EntityId entityId, DeviceEventTypeEnum attributeType,String eventIdentifier, String eventData,Long eventTime);
 
-    List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds);
 }

@@ -18,7 +18,7 @@ package org.thingsboard.server.dao.yunteng.jpa.entity.events;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 
 import javax.persistence.Column;
 import javax.persistence.Embeddable;
@@ -36,13 +36,11 @@ import static org.thingsboard.server.dao.model.ModelConstants.*;
 public class TkEventKvCompositeKey implements Serializable {
     @Enumerated(EnumType.STRING)
     @Column(name = EVENT_TYPE_COLUMN)
-    private TkEventType eventType;
+    private DeviceEventTypeEnum eventType;
     @Column(name = ENTITY_ID_COLUMN, columnDefinition = "uuid")
     private UUID entityId;
     @Column(name = EVENT_IDENTIFIER_COLUMN)
     private String eventIdentifier;
     @Column(name = EVENT_TIME_COLUMN)
     private Long eventTime;
-    @Column(name = EVENT_KEY_COLUMN)
-    private String eventKey;
 }

@@ -16,9 +16,8 @@
 package org.thingsboard.server.dao.yunteng.jpa.entity.events;
 
 import lombok.Data;
-import org.thingsboard.server.common.data.kv.*;
+import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvDto;
-import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
 import org.thingsboard.server.dao.model.ToData;
 
 import javax.persistence.Column;
@@ -28,50 +27,26 @@ import javax.persistence.Table;
 import java.io.Serializable;
 import java.util.UUID;
 
-import static org.thingsboard.server.dao.model.ModelConstants.*;
+import static org.thingsboard.server.dao.model.ModelConstants.DEVICE_DEVICE_PROFILE_ID_PROPERTY;
+import static org.thingsboard.server.dao.model.ModelConstants.EVENT_VALUE_COLUMN;
 
 @Data
 @Entity
 @Table(name = "tk_event_kv")
-public class TkEventKvEntity implements ToData<TkEventKvEntry>, Serializable {
+public class TkEventKvEntity implements ToData<TkEventKvDto>, Serializable {
 
     @EmbeddedId
     private TkEventKvCompositeKey id;
 
-    @Column(name = BOOLEAN_VALUE_COLUMN)
-    private Boolean booleanValue;
-
-    @Column(name = STRING_VALUE_COLUMN)
-    private String strValue;
-
-    @Column(name = LONG_VALUE_COLUMN)
-    private Long longValue;
-
-    @Column(name = DOUBLE_VALUE_COLUMN)
-    private Double doubleValue;
-
-    @Column(name = JSON_VALUE_COLUMN)
-    private String jsonValue;
+    @Column(name = EVENT_VALUE_COLUMN)
+    private String eventValue;
 
 
     @Column(name = DEVICE_DEVICE_PROFILE_ID_PROPERTY, columnDefinition = "uuid")
     private UUID deviceProfileId;
 
     @Override
-    public TkEventKvEntry toData() {
-        KvEntry kvEntry = null;
-        if (strValue != null) {
-            kvEntry = new StringDataEntry(id.getEventIdentifier(), strValue);
-        } else if (booleanValue != null) {
-            kvEntry = new BooleanDataEntry(id.getEventIdentifier(), booleanValue);
-        } else if (doubleValue != null) {
-            kvEntry = new DoubleDataEntry(id.getEventIdentifier(), doubleValue);
-        } else if (longValue != null) {
-            kvEntry = new LongDataEntry(id.getEventIdentifier(), longValue);
-        } else if (jsonValue != null) {
-            kvEntry = new JsonDataEntry(id.getEventIdentifier(), jsonValue);
-        }
-
-        return new TkEventKvDto(kvEntry, id.getEventTime());
+    public TkEventKvDto toData() {
+        return new TkEventKvDto(id.getEventTime(),id.getEventType(), JacksonUtil.toJsonNode(eventValue),id.getEntityId(),id.getEventIdentifier(),"",deviceProfileId,"");
     }
 }

@@ -13,7 +13,6 @@ import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvEntity;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
-import java.sql.Types;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.regex.Pattern;
@@ -25,14 +24,14 @@ public abstract class EventKvInsertRepository {
     private static final ThreadLocal<Pattern> PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE)));
     private static final String EMPTY_STR = "";
 
-    private static final String BATCH_UPDATE = "UPDATE tk_event_kv SET str_v = ?, long_v = ?, dbl_v = ?, bool_v = ?, json_v =  cast(? AS json) " +
-            "WHERE   event_key = ? and event_type = ? and entity_id = ? and event_identifier =? and event_time = ?;";
+    private static final String BATCH_UPDATE = "UPDATE tk_event_kv SET event_value = ?, device_profile_id = ?  " +
+            "WHERE    event_type = ? and entity_id = ? and event_identifier =? and event_time = ?;";
 
     private static final String INSERT_OR_UPDATE =
-            "INSERT INTO tk_event_kv (event_type, entity_id, event_identifier, event_time, str_v, long_v, dbl_v, bool_v, json_v, event_key) " +
-                    "VALUES(?, ?, ?, ?, ?, ?, ?, ?,  cast(? AS json), ?) " +
-                    "ON CONFLICT (event_type, entity_id, event_identifier, event_time, event_key) " +
-                    "DO UPDATE SET str_v = ?, long_v = ?, dbl_v = ?, bool_v = ?, json_v =  cast(? AS json), event_key = ?;";
+            "INSERT INTO tk_event_kv (event_type, entity_id, event_identifier, event_time, event_value,device_profile_id) " +
+                    "VALUES(?, ?, ?, ?, ?, ?) " +
+                    "ON CONFLICT (event_type, entity_id, event_identifier, event_time) " +
+                    "DO UPDATE SET event_value = ?,device_profile_id = ?;";
 
     @Autowired
     protected JdbcTemplate jdbcTemplate;
@@ -51,33 +50,12 @@ public abstract class EventKvInsertRepository {
                     @Override
                     public void setValues(PreparedStatement ps, int i) throws SQLException {
                         TkEventKvEntity kvEntity = entities.get(i);
-                        ps.setString(1, replaceNullChars(kvEntity.getStrValue()));
-
-                        if (kvEntity.getLongValue() != null) {
-                            ps.setLong(2, kvEntity.getLongValue());
-                        } else {
-                            ps.setNull(2, Types.BIGINT);
-                        }
-
-                        if (kvEntity.getDoubleValue() != null) {
-                            ps.setDouble(3, kvEntity.getDoubleValue());
-                        } else {
-                            ps.setNull(3, Types.DOUBLE);
-                        }
-
-                        if (kvEntity.getBooleanValue() != null) {
-                            ps.setBoolean(4, kvEntity.getBooleanValue());
-                        } else {
-                            ps.setNull(4, Types.BOOLEAN);
-                        }
-
-                        ps.setString(5, replaceNullChars(kvEntity.getJsonValue()));
-
-                        ps.setString(6, kvEntity.getId().getEventKey());
-                        ps.setString(7, kvEntity.getId().getEventType().name());
-                        ps.setObject(8, kvEntity.getId().getEntityId());
-                        ps.setString(9, kvEntity.getId().getEventIdentifier());
-                        ps.setLong(10, kvEntity.getId().getEventTime());
+                        ps.setString(1, replaceNullChars(kvEntity.getEventValue()));
+                        ps.setObject(2, kvEntity.getDeviceProfileId());
+                        ps.setString(3, kvEntity.getId().getEventType().name());
+                        ps.setObject(4, kvEntity.getId().getEntityId());
+                        ps.setString(5, kvEntity.getId().getEventIdentifier());
+                        ps.setLong(6, kvEntity.getId().getEventTime());
                     }
 
                     @Override
@@ -109,38 +87,12 @@ public abstract class EventKvInsertRepository {
                         ps.setString(3, kvEntity.getId().getEventIdentifier());
                         ps.setLong(4, kvEntity.getId().getEventTime());
 
-                        ps.setString(5, replaceNullChars(kvEntity.getStrValue()));
-                        ps.setString(11, replaceNullChars(kvEntity.getStrValue()));
-
-                        if (kvEntity.getLongValue() != null) {
-                            ps.setLong(6, kvEntity.getLongValue());
-                            ps.setLong(12, kvEntity.getLongValue());
-                        } else {
-                            ps.setNull(6, Types.BIGINT);
-                            ps.setNull(12, Types.BIGINT);
-                        }
-
-                        if (kvEntity.getDoubleValue() != null) {
-                            ps.setDouble(7, kvEntity.getDoubleValue());
-                            ps.setDouble(13, kvEntity.getDoubleValue());
-                        } else {
-                            ps.setNull(7, Types.DOUBLE);
-                            ps.setNull(13, Types.DOUBLE);
-                        }
-
-                        if (kvEntity.getBooleanValue() != null) {
-                            ps.setBoolean(8, kvEntity.getBooleanValue());
-                            ps.setBoolean(14, kvEntity.getBooleanValue());
-                        } else {
-                            ps.setNull(8, Types.BOOLEAN);
-                            ps.setNull(14, Types.BOOLEAN);
-                        }
-
-                        ps.setString(9, replaceNullChars(kvEntity.getJsonValue()));
-                        ps.setString(15, replaceNullChars(kvEntity.getJsonValue()));
-
-                        ps.setString(10, kvEntity.getId().getEventKey());
-                        ps.setString(16, kvEntity.getId().getEventKey());
+                        ps.setString(5, replaceNullChars(kvEntity.getEventValue()));
+                        ps.setString(7, replaceNullChars(kvEntity.getEventValue()));
+
+
+                        ps.setObject(6, kvEntity.getDeviceProfileId());
+                        ps.setObject(8, kvEntity.getDeviceProfileId());
                     }
 
                     @Override

@@ -15,47 +15,22 @@
  */
 package org.thingsboard.server.dao.yunteng.jpa.repository.event;
 
-import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.Query;
-import org.springframework.data.repository.CrudRepository;
+import org.springframework.data.repository.PagingAndSortingRepository;
 import org.springframework.data.repository.query.Param;
-import org.springframework.transaction.annotation.Transactional;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
 import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvCompositeKey;
 import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvEntity;
 
-import java.util.List;
 import java.util.UUID;
 
-public interface EventKvRepository extends CrudRepository<TkEventKvEntity, TkEventKvCompositeKey> {
+public interface EventKvRepository extends PagingAndSortingRepository<TkEventKvEntity, TkEventKvCompositeKey> {
 
-    @Query("SELECT a FROM TkEventKvEntity a WHERE a.id.eventType = :eventType " +
-            "AND a.id.entityId = :entityId " )
-    List<TkEventKvEntity> findAllByEventTypeAndEntityId(@Param("eventType") TkEventType eventType,
-                                                                           @Param("entityId") UUID entityId);
-
-    @Transactional
-    @Modifying
-    @Query("DELETE FROM TkEventKvEntity a WHERE a.id.eventType = :eventType " +
-            "AND a.id.entityId = :entityId " +
-            "AND a.id.eventIdentifier = :eventIdentifier " +
-            "AND a.id.eventKey = :eventKey")
-    void delete(@Param("eventType") TkEventType eventType,
-                @Param("entityId") UUID entityId,
-                @Param("eventIdentifier") String eventIdentifier,
-                @Param("eventKey") String eventKey);
-
-    @Query(value = "SELECT DISTINCT event_key FROM tk_event_kv WHERE entity_id " +
-            " in (SELECT id FROM device WHERE tenant_id = :tenantId and device_profile_id = :deviceProfileId limit 100) ORDER BY event_key", nativeQuery = true)
-    List<String> findAllKeysByDeviceProfileId(@Param("tenantId") UUID tenantId,
-                                              @Param("deviceProfileId") UUID deviceProfileId);
-
-    @Query(value = "SELECT DISTINCT event_key FROM tk_event_kv WHERE entity_id  " +
-            "  in (SELECT id FROM device WHERE tenant_id = :tenantId limit 100) ORDER BY event_key", nativeQuery = true)
-    List<String> findAllKeysByTenantId(@Param("tenantId") UUID tenantId);
-
-    @Query(value = "SELECT DISTINCT event_key FROM tk_event_kv WHERE entity_id " +
-            " in :entityIds ORDER BY event_key", nativeQuery = true)
-    List<String> findAllKeysByEntityIds(@Param("entityIds") List<UUID> entityIds);
+    @Query("SELECT a FROM TkEventKvEntity a WHERE a.id.entityId = :entityId "
+//            +"AND a.id.eventType = :eventType AND a.id.eventIdentifier = :eventIdentifier "
+    )
+    Page<TkEventKvEntity> findEvents(@Param("entityId") UUID entityId,
+                                     Pageable pageable);
 }
 

@@ -29,16 +29,15 @@ import java.util.List;
 public class HsqlEventsInsertRepository extends EventKvInsertRepository {
 
     private static final String INSERT_OR_UPDATE =
-            "MERGE INTO tk_event_kv USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " +
-                    "A (event_type, entity_id, event_identifier, event_key, str_v, long_v, dbl_v, bool_v, json_v, event_key) " +
+            "MERGE INTO tk_event_kv USING(VALUES ?, ?, ?, ?, ?, ?) " +
+                    "A (event_type, entity_id, event_identifier, event_time, event_value,device_profile_id) " +
                     "ON (tk_event_kv.event_type=A.event_type " +
                     "AND tk_event_kv.entity_id=A.entity_id " +
                     "AND tk_event_kv.event_identifier=A.event_identifier " +
                     "AND tk_event_kv.event_time=A.event_time) " +
-                    "AND tk_event_kv.event_key=A.event_key) " +
-                    "WHEN MATCHED THEN UPDATE SET tk_event_kv.str_v = A.str_v, tk_event_kv.long_v = A.long_v, tk_event_kv.dbl_v = A.dbl_v, tk_event_kv.bool_v = A.bool_v, tk_event_kv.json_v = A.json_v " +
-                    "WHEN NOT MATCHED THEN INSERT (event_type, entity_id, event_identifier, event_time, str_v, long_v, dbl_v, bool_v, json_v, event_key) " +
-                    "VALUES (A.event_type, A.entity_id, A.event_identifier, A.event_time, A.str_v, A.long_v, A.dbl_v, A.bool_v, A.json_v, A.event_key)";
+                    "WHEN MATCHED THEN UPDATE SET tk_event_kv.event_value = A.event_value " +
+                    "WHEN NOT MATCHED THEN INSERT (event_type, entity_id, event_identifier, event_time, event_value, device_profile_id) " +
+                    "VALUES (A.event_type, A.entity_id, A.event_identifier, A.event_time, A.event_value, A.device_profile_id)";
 
     @Override
     protected void saveOrUpdate(List<TkEventKvEntity> entities) {
@@ -48,29 +47,8 @@ public class HsqlEventsInsertRepository extends EventKvInsertRepository {
                 ps.setObject(2, entity.getId().getEntityId());
                 ps.setString(3, entity.getId().getEventIdentifier());
                 ps.setLong(4, entity.getId().getEventTime());
-                ps.setString(5, entity.getStrValue());
-
-                if (entity.getLongValue() != null) {
-                    ps.setLong(6, entity.getLongValue());
-                } else {
-                    ps.setNull(6, Types.BIGINT);
-                }
-
-                if (entity.getDoubleValue() != null) {
-                    ps.setDouble(7, entity.getDoubleValue());
-                } else {
-                    ps.setNull(7, Types.DOUBLE);
-                }
-
-                if (entity.getBooleanValue() != null) {
-                    ps.setBoolean(8, entity.getBooleanValue());
-                } else {
-                    ps.setNull(8, Types.BOOLEAN);
-                }
-
-                ps.setString(9, entity.getJsonValue());
-
-                ps.setString(10, entity.getId().getEventKey());
+                ps.setObject(5, entity.getEventValue());
+                ps.setObject(6, entity.getEventValue());
             });
         });
     }

@@ -15,37 +15,32 @@
  */
 package org.thingsboard.server.dao.yunteng.jpa.repository.event;
 
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.data.EntityType;
 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.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+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.enums.DeviceEventTypeEnum;
 import org.thingsboard.server.common.stats.StatsFactory;
 import org.thingsboard.server.dao.DaoUtil;
 import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService;
 import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
+import org.thingsboard.server.dao.yunteng.jpa.dao.event.EventsDao;
 import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvCompositeKey;
 import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvEntity;
-import org.thingsboard.server.dao.yunteng.jpa.dao.event.EventsDao;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.util.Collection;
 import java.util.Comparator;
-import java.util.List;
-import java.util.Optional;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 
 @Component
 @Slf4j
@@ -109,60 +104,18 @@ public class JpaEventDao extends JpaAbstractDaoListeningExecutorService implemen
     }
 
     @Override
-    public ListenableFuture<Optional<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId, TkEventType eventType, String attributeKey) {
-        TkEventKvCompositeKey compositeKey =
-                getEventKvCompositeKey(eventType,entityId,null,  attributeKey,null);
-        return Futures.immediateFuture(
-                Optional.ofNullable(DaoUtil.getData(eventKvRepository.findById(compositeKey))));
-    }
+    public PageData<TkEventKvDto> findEvents(EntityId entityId,String eventIdentifier,  DeviceEventTypeEnum eventType, TimePageLink pageLink) {
 
-    @Override
-    public ListenableFuture<List<TkEventKvEntry>> find(TenantId tenantId, EntityId entityId, TkEventType eventType, Collection<String> attributeKeys) {
-        List<TkEventKvCompositeKey> compositeKeys =
-                attributeKeys
-                        .stream()
-                        .map(attributeKey ->
-                                getEventKvCompositeKey(eventType,entityId,  null,attributeKey,null))
-                        .collect(Collectors.toList());
-        return Futures.immediateFuture(
-                DaoUtil.convertDataList(Lists.newArrayList(eventKvRepository.findAllById(compositeKeys))));
+        return DaoUtil.toPageData(eventKvRepository.findEvents(entityId.getId(),DaoUtil.toPageable(pageLink)));
     }
 
-    @Override
-    public ListenableFuture<List<TkEventKvEntry>> findAll(TenantId tenantId, EntityId entityId, TkEventType eventType) {
-        return Futures.immediateFuture(
-                DaoUtil.convertDataList(Lists.newArrayList(
-                        eventKvRepository.findAllByEventTypeAndEntityId(
-                                eventType,
-                                entityId.getId()))));
-    }
 
     @Override
-    public List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId) {
-        if (deviceProfileId != null) {
-            return eventKvRepository.findAllKeysByDeviceProfileId(tenantId.getId(), deviceProfileId.getId());
-        } else {
-            return eventKvRepository.findAllKeysByTenantId(tenantId.getId());
-        }
-    }
-
-    @Override
-    public List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds) {
-        return eventKvRepository
-                .findAllKeysByEntityIds(entityIds.stream().map(EntityId::getId).collect(Collectors.toList()));
-    }
-
-    @Override
-    public ListenableFuture<Void> save(TenantId tenantId, EntityId entityId,TkEventType eventType, String eventIdentifier, TkEventKvEntry attribute) {
+    public ListenableFuture<Void> save(TenantId tenantId, DeviceProfileId profileId,EntityId entityId,DeviceEventTypeEnum eventType, String eventIdentifier, String eventData,Long eventTime) {
         TkEventKvEntity entity = new TkEventKvEntity();
-        entity.setId(new TkEventKvCompositeKey(eventType, entityId.getId(), eventIdentifier, attribute.getEventTime(),attribute.getKey()));
-
-//        entity.setDeviceProfileId();
-        entity.setStrValue(attribute.getStrValue().orElse(null));
-        entity.setDoubleValue(attribute.getDoubleValue().orElse(null));
-        entity.setLongValue(attribute.getLongValue().orElse(null));
-        entity.setBooleanValue(attribute.getBooleanValue().orElse(null));
-        entity.setJsonValue(attribute.getJsonValue().orElse(null));
+        entity.setId(new TkEventKvCompositeKey(eventType, entityId.getId(), eventIdentifier, eventTime));
+        entity.setDeviceProfileId(profileId.getId());
+        entity.setEventValue(eventData);
         return addToQueue(entity);
     }
 
@@ -170,22 +123,4 @@ public class JpaEventDao extends JpaAbstractDaoListeningExecutorService implemen
         return queue.add(entity);
     }
 
-    @Override
-    public ListenableFuture<List<Void>> removeAll(TenantId tenantId, EntityId entityId, String eventId,TkEventType eventType,List<String> keys) {
-        return service.submit(() -> {
-            keys.forEach(key ->
-                    eventKvRepository.delete(eventType,entityId.getId(), eventId,key)
-            );
-            return null;
-        });
-    }
-
-    private TkEventKvCompositeKey getEventKvCompositeKey(TkEventType eventType,EntityId entityId, String eventIdentifier,String eventKey, Long eventTime) {
-        return new TkEventKvCompositeKey(
-                eventType,
-                entityId.getId(),
-                eventIdentifier,
-                eventTime,
-                eventKey);
-    }
 }

@@ -17,13 +17,14 @@ package org.thingsboard.rule.engine.api;
 
 import com.google.common.util.concurrent.FutureCallback;
 import org.thingsboard.server.common.data.id.CustomerId;
+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.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 
 import java.util.Collection;
 import java.util.List;
@@ -44,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, EntityId entityId, String eventId, TkEventType eventType, List<TkEventKvEntry> attributes, FutureCallback<Void> callback);
+    void saveAndNotify(TenantId tenantId, DeviceProfileId profileId, EntityId entityId, String eventId, DeviceEventTypeEnum eventType, String eventData,Long eventTime, FutureCallback<Void> callback);
 
     void saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 

@@ -15,25 +15,19 @@
  */
 package org.thingsboard.rule.engine.yunteng.event;
 
-import com.google.gson.JsonParser;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 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.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.plugin.ComponentType;
-import org.thingsboard.server.common.data.yunteng.dto.TkEventKvEntry;
-import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceEventTypeEnum;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.session.SessionMsgType;
-import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 
-import java.util.ArrayList;
-import java.util.Set;
+import java.util.UUID;
 
-import static org.thingsboard.server.dao.model.ModelConstants.EVENT_IDENTIFIER_COLUMN;
-import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TYPE_COLUMN;
+import static org.thingsboard.server.dao.model.ModelConstants.*;
 
 @Slf4j
 @RuleNode(
@@ -49,10 +43,12 @@ import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TYPE_COLUMN;
 public class TkMsgEventNode implements TbNode {
 
     private TkMsgEventNodeConfiguration config;
+    private RuleEngineDeviceProfileCache cache;
 
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TkMsgEventNodeConfiguration.class);
+        cache = ctx.getDeviceProfileCache();
     }
 
     @Override
@@ -61,16 +57,18 @@ public class TkMsgEventNode implements TbNode {
             ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType()));
             return;
         }
-        String src = msg.getData();
-        Set<TkEventKvEntry> events = JsonConverter.convertToEvents(new JsonParser().parse(src));
         String eventIdentifier = msg.getMetaData().getValue(EVENT_IDENTIFIER_COLUMN);
-        TkEventType eventType = TkEventType.valueOf(msg.getMetaData().getValue(EVENT_TYPE_COLUMN));
+        String deviceProfileId = msg.getMetaData().getValue(DEVICE_DEVICE_PROFILE_ID_PROPERTY);
+        ////TODO: 验证事件类型、事件标识符和数据建是否与产品物模型中的事件匹配
+        long ts = System.currentTimeMillis();
+        String src = msg.getData();
+        DeviceEventTypeEnum eventType = DeviceEventTypeEnum.valueOf(msg.getMetaData().getValue(EVENT_TYPE_COLUMN));
         ctx.getTelemetryService().saveAndNotify(
-                ctx.getTenantId(),
+                ctx.getTenantId(),new DeviceProfileId(UUID.fromString(deviceProfileId)),
                 msg.getOriginator(),
                 eventIdentifier,
                 eventType,
-                new ArrayList<>(events),
+                src,ts,
                 new TelemetryNodeCallback(ctx, msg)
         );
     }