feat: 设备事件上报

@@ -39,12 +39,15 @@ 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.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.entityview.EntityViewService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.dao.yunteng.event.TkEventsService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.usagestats.TbApiUsageClient;
@@ -74,6 +77,8 @@ import java.util.concurrent.Executors;
 public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionService implements TelemetrySubscriptionService {
 
     private final AttributesService attrService;
+
+    private final TkEventsService eventsService;
     private final TimeseriesService tsService;
     private final EntityViewService entityViewService;
     private final TbApiUsageClient apiUsageClient;
@@ -82,6 +87,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
     private ExecutorService tsCallBackExecutor;
 
     public DefaultTelemetrySubscriptionService(AttributesService attrService,
+                                               TkEventsService eventsService,
                                                TimeseriesService tsService,
                                                EntityViewService entityViewService,
                                                TbClusterService clusterService,
@@ -90,6 +96,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
                                                TbApiUsageStateService apiUsageStateService) {
         super(clusterService, partitionService);
         this.attrService = attrService;
+        this.eventsService = eventsService;
         this.tsService = tsService;
         this.entityViewService = entityViewService;
         this.apiUsageClient = apiUsageClient;
@@ -241,6 +248,20 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
         saveAndNotifyInternal(tenantId, entityId, scope, attributes, notifyDevice, callback);
     }
 
+
+    //Thingskit function
+    @Override
+    public void saveAndNotify(TenantId tenantId, EntityId entityId, String eventId, TkEventType eventType, List<TkEventKvEntry> attributes, FutureCallback<Void> callback) {
+        checkInternalEntity(entityId);
+        saveAndNotifyInternal(tenantId, entityId,  eventId,eventType, attributes, 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);
+        addVoidCallback(saveFuture, callback);
+//        addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice));
+    }
+
     @Override
     public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback) {
         ListenableFuture<List<Void>> saveFuture = attrService.save(tenantId, entityId, scope, attributes);

@@ -21,6 +21,8 @@ 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 java.util.List;
 
@@ -35,6 +37,9 @@ 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 saveLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 
     void deleteAndNotifyInternal(TenantId tenantId, EntityId entityId, String scope, List<String> keys, FutureCallback<Void> callback);

+/**
+ * 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.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 java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * @author Andrew Shvayka
+ */
+public interface TkEventsService {
+
+    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);
+
+}

+/**
+ * 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.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.DataType;
+import org.thingsboard.server.common.data.kv.KvEntry;
+
+import javax.validation.Valid;
+import java.util.Optional;
+
+/**
+ * @author Andrew Shvayka
+ */
+public class TkEventKvDto implements TkEventKvEntry {
+
+    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 +
+                '}';
+    }
+}

+/**
+ * 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;
+
+/**
+ * @author Andrew Shvayka
+ */
+public interface TkEventKvEntry extends KvEntry {
+
+    long getEventTime();
+
+}

@@ -34,6 +34,8 @@ 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.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
@@ -516,6 +518,13 @@ 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<>();
         for (Entry<String, JsonElement> valueEntry : valuesObject.entrySet()) {

@@ -293,8 +293,8 @@ public class DefaultTransportService implements TransportService {
 
     @Override
     public TransportProtos.GetEntityProfileResponseMsg getEntityProfile(TransportProtos.GetEntityProfileRequestMsg msg) {
-        TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
-                new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setEntityProfileRequestMsg(msg).build());
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg =
+                new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setEntityProfileRequestMsg(msg).build());
         try {
             TbProtoQueueMsg<TransportApiResponseMsg> response = transportApiRequestTemplate.send(protoMsg).get();
             return response.getValue().getEntityProfileResponseMsg();
@@ -305,8 +305,8 @@ public class DefaultTransportService implements TransportService {
 
     @Override
     public TransportProtos.GetResourceResponseMsg getResource(TransportProtos.GetResourceRequestMsg msg) {
-        TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
-                new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setResourceRequestMsg(msg).build());
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg =
+                new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setResourceRequestMsg(msg).build());
         try {
             TbProtoQueueMsg<TransportApiResponseMsg> response = transportApiRequestTemplate.send(protoMsg).get();
             return response.getValue().getResourceResponseMsg();
@@ -317,8 +317,8 @@ public class DefaultTransportService implements TransportService {
 
     @Override
     public TransportProtos.GetSnmpDevicesResponseMsg getSnmpDevicesIds(TransportProtos.GetSnmpDevicesRequestMsg requestMsg) {
-        TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(
-                UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder()
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(
+                UUID.randomUUID(), TransportApiRequestMsg.newBuilder()
                 .setSnmpDevicesRequestMsg(requestMsg)
                 .build()
         );
@@ -334,7 +334,7 @@ public class DefaultTransportService implements TransportService {
     @Override
     public TransportProtos.GetDeviceResponseMsg getDevice(TransportProtos.GetDeviceRequestMsg requestMsg) {
         TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(
-                UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder()
+                UUID.randomUUID(), TransportApiRequestMsg.newBuilder()
                 .setDeviceRequestMsg(requestMsg)
                 .build()
         );
@@ -354,7 +354,7 @@ public class DefaultTransportService implements TransportService {
     @Override
     public TransportProtos.GetDeviceCredentialsResponseMsg getDeviceCredentials(TransportProtos.GetDeviceCredentialsRequestMsg requestMsg) {
         TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(
-                UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder()
+                UUID.randomUUID(), TransportApiRequestMsg.newBuilder()
                 .setDeviceCredentialsRequestMsg(requestMsg)
                 .build()
         );
@@ -576,25 +576,26 @@ public class DefaultTransportService implements TransportService {
     @Override
     public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.PostEventMsg msg,
                         TransportServiceCallback<Void> callback,String topicName) {
-        //Topic地址：v1/devices/event/{${identifier}}/{$eventType}
-        String[] topicInfo = topicName.split(MqttTopics.BASE_DEVICE_EVENT_TOPIC);
+        //Topic地址：v1/devices/event/${deviceId}/${identifier}/{$eventType}
+        String[] topicInfo = topicName.split(MqttTopics.BASE_DEVICE_EVENT_TOPIC + "/");
         if(null == topicInfo || topicInfo.length !=2){
             throw new TkDataValidationException(ErrorMessage.INVALID_TOPIC.getMessage());
         }
         String[] eventInfo = topicInfo[1].split("/");
-        if(null == eventInfo || eventInfo.length !=2){
+        if(null == eventInfo || eventInfo.length !=3){
             throw new TkDataValidationException(ErrorMessage.INVALID_TOPIC.getMessage());
         }
         reportActivityInternal(sessionInfo);
         JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
         TbMsgMetaData metaData = new TbMsgMetaData();
-        metaData.putValue("deviceName", sessionInfo.getDeviceName());
-        metaData.putValue("deviceType", sessionInfo.getDeviceType());
-        metaData.putValue("event_identifier", eventInfo[0]);
-        metaData.putValue("event_type",eventInfo[1]);
         CustomerId customerId = getCustomerId(sessionInfo);
         TenantId tenantId = getTenantId(sessionInfo);
         DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()));
+        metaData.putValue("deviceName", sessionInfo.getDeviceName());
+        metaData.putValue("deviceType", sessionInfo.getDeviceType());
+        metaData.putValue("deviceId",eventInfo[0]);
+        metaData.putValue("event_identifier", eventInfo[1]);
+        metaData.putValue("event_type",eventInfo[2]);
         sendToRuleEngine(tenantId, deviceId, customerId, sessionInfo, json, metaData, SessionMsgType.POST_EVENT_REQUEST,
                 new TransportTbQueueCallback(new ApiStatsProxyCallback<>(tenantId, customerId, msg.getKvList().size(), callback)));
     }
@@ -726,8 +727,8 @@ public class DefaultTransportService implements TransportService {
     @Override
     public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.GetOtaPackageRequestMsg msg, TransportServiceCallback<TransportProtos.GetOtaPackageResponseMsg> callback) {
         if (checkLimits(sessionInfo, msg, callback)) {
-            TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
-                    new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setOtaPackageRequestMsg(msg).build());
+            TbProtoQueueMsg<TransportApiRequestMsg> protoMsg =
+                    new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setOtaPackageRequestMsg(msg).build());
 
             AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg), response -> {
                 callback.onSuccess(response.getValue().getOtaPackageResponseMsg());
@@ -870,7 +871,7 @@ public class DefaultTransportService implements TransportService {
         }
     }
 
-    protected void processToTransportMsg(TransportProtos.ToTransportMsg toSessionMsg) {
+    protected void processToTransportMsg(ToTransportMsg toSessionMsg) {
         UUID sessionId = new UUID(toSessionMsg.getSessionIdMSB(), toSessionMsg.getSessionIdLSB());
         SessionMetaData md = sessions.get(sessionId);
         if (md != null) {

@@ -56,6 +56,12 @@ public class ModelConstants {
     public static final String ATTRIBUTE_KEY_COLUMN = "attribute_key";
     public static final String LAST_UPDATE_TS_COLUMN = "last_update_ts";
 
+    //Thingskit function
+    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";
+
     /**
      * Cassandra user constants.
      */

+/**
+ * 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.dao.yunteng.jpa.dao.event;
+
+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.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;
+
+
+/**
+ * @author Andrew Shvayka
+ */
+@Service
+@Slf4j
+public class BaseEventsService implements TkEventsService {
+    private final EventsDao eventsDao;
+
+    public BaseEventsService(EventsDao eventsDao) {
+        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);
+    }
+
+    @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) {
+        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());
+        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);
+    }
+}

+/**
+ * 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.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.dao.exception.IncorrectParameterException;
+import org.thingsboard.server.dao.service.Validator;
+
+public class EventUtils {
+    public static void validate(EntityId id, String eventIdentifier) {
+        Validator.validateId(id.getId(), "Incorrect id " + id);
+        Validator.validateString(eventIdentifier, "Incorrect scope " + eventIdentifier);
+    }
+
+
+    public static void validate(EntityId id, String eventIdentifier, TkEventType eventType) {
+        Validator.validateId(id.getId(), "Incorrect id " + id);
+        Validator.validateString(eventIdentifier, "Incorrect scope " + eventIdentifier);
+        if(eventType == null){
+            throw new IncorrectParameterException("eventType entry can't be null");
+        }
+    }
+
+    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");
+        }
+    }
+}

+/**
+ * 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.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;
+
+/**
+ * @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);
+
+    List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId);
+
+    List<String> findAllKeysByEntityIds(TenantId tenantId, EntityType entityType, List<EntityId> entityIds);
+}

+/**
+ * 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.dao.yunteng.jpa.entity.events;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.thingsboard.server.common.data.yunteng.enums.TkEventType;
+
+import javax.persistence.Column;
+import javax.persistence.Embeddable;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
+import java.io.Serializable;
+import java.util.UUID;
+
+import static org.thingsboard.server.dao.model.ModelConstants.*;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Embeddable
+public class TkEventKvCompositeKey implements Serializable {
+    @Enumerated(EnumType.STRING)
+    @Column(name = EVENT_TYPE_COLUMN)
+    private TkEventType 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;
+}

+/**
+ * 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.dao.yunteng.jpa.entity.events;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.kv.*;
+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;
+import javax.persistence.EmbeddedId;
+import javax.persistence.Entity;
+import javax.persistence.Table;
+import java.io.Serializable;
+import java.util.UUID;
+
+import static org.thingsboard.server.dao.model.ModelConstants.*;
+
+@Data
+@Entity
+@Table(name = "tk_event_kv")
+public class TkEventKvEntity implements ToData<TkEventKvEntry>, 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 = 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());
+    }
+}

+package org.thingsboard.server.dao.yunteng.jpa.repository.event;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.BatchPreparedStatementSetter;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
+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;
+
+@Repository
+@Slf4j
+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 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 = ?;";
+
+    @Autowired
+    protected JdbcTemplate jdbcTemplate;
+
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
+    @Value("${sql.remove_null_chars:true}")
+    private boolean removeNullChars;
+
+    protected void saveOrUpdate(List<TkEventKvEntity> entities) {
+        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+            @Override
+            protected void doInTransactionWithoutResult(TransactionStatus status) {
+                int[] result = jdbcTemplate.batchUpdate(BATCH_UPDATE, new BatchPreparedStatementSetter() {
+                    @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());
+                    }
+
+                    @Override
+                    public int getBatchSize() {
+                        return entities.size();
+                    }
+                });
+
+                int updatedCount = 0;
+                for (int i = 0; i < result.length; i++) {
+                    if (result[i] == 0) {
+                        updatedCount++;
+                    }
+                }
+
+                List<TkEventKvEntity> insertEntities = new ArrayList<>(updatedCount);
+                for (int i = 0; i < result.length; i++) {
+                    if (result[i] == 0) {
+                        insertEntities.add(entities.get(i));
+                    }
+                }
+
+                jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() {
+                    @Override
+                    public void setValues(PreparedStatement ps, int i) throws SQLException {
+                        TkEventKvEntity kvEntity = insertEntities.get(i);
+                        ps.setString(1, kvEntity.getId().getEventType().name());
+                        ps.setObject(2, kvEntity.getId().getEntityId());
+                        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());
+                    }
+
+                    @Override
+                    public int getBatchSize() {
+                        return insertEntities.size();
+                    }
+                });
+            }
+        });
+    }
+
+    private String replaceNullChars(String strValue) {
+        if (removeNullChars && strValue != null) {
+            return PATTERN_THREAD_LOCAL.get().matcher(strValue).replaceAll(EMPTY_STR);
+        }
+        return strValue;
+    }
+}

+/**
+ * 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.dao.yunteng.jpa.repository.event;
+
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.CrudRepository;
+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> {
+
+    @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);
+}
+

+/**
+ * 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.dao.yunteng.jpa.repository.event;
+
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+import org.thingsboard.server.dao.util.HsqlDao;
+import org.thingsboard.server.dao.yunteng.jpa.entity.events.TkEventKvEntity;
+
+import java.sql.Types;
+import java.util.List;
+
+@HsqlDao
+@Repository
+@Transactional
+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) " +
+                    "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)";
+
+    @Override
+    protected void saveOrUpdate(List<TkEventKvEntity> entities) {
+        entities.forEach(entity -> {
+            jdbcTemplate.update(INSERT_OR_UPDATE, ps -> {
+                ps.setString(1, entity.getId().getEventType().name());
+                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());
+            });
+        });
+    }
+}

+/**
+ * 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.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.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.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
+public class JpaEventDao extends JpaAbstractDaoListeningExecutorService implements EventsDao {
+
+    @Autowired
+    ScheduledLogExecutorComponent logExecutor;
+
+    @Autowired
+    private EventKvRepository eventKvRepository;
+
+    @Autowired
+    private EventKvInsertRepository eventKvInsertRepository;
+
+    @Autowired
+    private StatsFactory statsFactory;
+
+    @Value("${sql.attributes.batch_size:1000}")
+    private int batchSize;
+
+    @Value("${sql.attributes.batch_max_delay:100}")
+    private long maxDelay;
+
+    @Value("${sql.attributes.stats_print_interval_ms:1000}")
+    private long statsPrintIntervalMs;
+
+    @Value("${sql.attributes.batch_threads:4}")
+    private int batchThreads;
+
+    @Value("${sql.batch_sort:false}")
+    private boolean batchSortEnabled;
+
+    private TbSqlBlockingQueueWrapper<TkEventKvEntity> queue;
+
+    @PostConstruct
+    private void init() {
+        TbSqlBlockingQueueParams params = TbSqlBlockingQueueParams.builder()
+                .logName("Events")
+                .batchSize(batchSize)
+                .maxDelay(maxDelay)
+                .statsPrintIntervalMs(statsPrintIntervalMs)
+                .statsNamePrefix("events")
+                .batchSortEnabled(batchSortEnabled)
+                .build();
+
+        Function<TkEventKvEntity, Integer> hashcodeFunction = entity -> entity.getId().getEntityId().hashCode();
+        queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory);
+        queue.init(logExecutor, v -> eventKvInsertRepository.saveOrUpdate(v),
+                Comparator.comparing((TkEventKvEntity attributeKvEntity) -> attributeKvEntity.getId().getEntityId())
+                        .thenComparing(attributeKvEntity -> attributeKvEntity.getId().getEventType().name())
+                        .thenComparing(attributeKvEntity -> attributeKvEntity.getId().getEventTime())
+                        .thenComparing(attributeKvEntity -> attributeKvEntity.getId().getEventIdentifier())
+        );
+    }
+
+    @PreDestroy
+    private void destroy() {
+        if (queue != null) {
+            queue.destroy();
+        }
+    }
+
+    @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))));
+    }
+
+    @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))));
+    }
+
+    @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) {
+        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));
+        return addToQueue(entity);
+    }
+
+    private ListenableFuture<Void> addToQueue(TkEventKvEntity entity) {
+        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);
+    }
+}

+/**
+ * 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.dao.yunteng.jpa.repository.event;
+
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+import org.thingsboard.server.dao.util.PsqlDao;
+
+@PsqlDao
+@Repository
+@Transactional
+public class PsqlEventsInsertRepository extends EventKvInsertRepository {
+
+}

@@ -22,6 +22,8 @@ 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 java.util.Collection;
 import java.util.List;
@@ -41,6 +43,9 @@ 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 saveLatestAndNotify(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Void> callback);
 
     void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback);

@@ -33,7 +33,7 @@ import org.thingsboard.server.common.msg.session.SessionMsgType;
         type = ComponentType.FILTER,
         name = "message type switch",
         configClazz = EmptyNodeConfiguration.class,
-        relationTypes = {"Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Sent", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Expired", "RPC Failed", "RPC Deleted",
+        relationTypes = {"Post event","Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Sent", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Expired", "RPC Failed", "RPC Deleted",
                 "Activity Event", "Inactivity Event", "Connect Event", "Disconnect Event", "Entity Created", "Entity Updated", "Entity Deleted", "Entity Assigned",
                 "Entity Unassigned", "Attributes Updated", "Attributes Deleted", "Alarm Acknowledged", "Alarm Cleared", "Other", "Entity Assigned From Tenant", "Entity Assigned To Tenant",
                 "Timeseries Updated", "Timeseries Deleted"},
@@ -57,7 +57,14 @@ public class TbMsgTypeSwitchNode implements TbNode {
             relationType = "Post attributes";
         } else if (msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) {
             relationType = "Post telemetry";
-        } else if (msg.getType().equals(SessionMsgType.TO_SERVER_RPC_REQUEST.name())) {
+        }
+
+        //Thingskit function
+        else if (msg.getType().equals(SessionMsgType.POST_EVENT_REQUEST.name())) {
+            relationType = "Post event";
+        }
+
+        else if (msg.getType().equals(SessionMsgType.TO_SERVER_RPC_REQUEST.name())) {
             relationType = "RPC Request from Device";
         } else if (msg.getType().equals(DataConstants.ACTIVITY_EVENT)) {
             relationType = "Activity Event";

@@ -26,6 +26,7 @@ import javax.annotation.Nullable;
  * Created by ashvayka on 02.04.18.
  */
 @Data
+public
 class TelemetryNodeCallback implements FutureCallback<Void> {
     private final TbContext ctx;
     private final TbMsg msg;

+/**
+ * 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.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.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.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 static org.thingsboard.server.dao.model.ModelConstants.EVENT_IDENTIFIER_COLUMN;
+import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TYPE_COLUMN;
+
+@Slf4j
+@RuleNode(
+        type = ComponentType.ACTION,
+        name = "save event",
+        configClazz = TkMsgEventNodeConfiguration.class,
+        nodeDescription = "Saves device event data",
+        nodeDetails = "Saves entity event . Expects messages with 'POST_EVENT_REQUEST' message type",
+        uiResources = {"static/rulenode/rulenode-core-config.js"},
+        configDirective = "tkMsgEventNodeConfiguration",
+        icon = "file_upload"
+)
+public class TkMsgEventNode implements TbNode {
+
+    private TkMsgEventNodeConfiguration config;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TkMsgEventNodeConfiguration.class);
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) {
+        if (!msg.getType().equals(SessionMsgType.POST_EVENT_REQUEST.name())) {
+            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));
+        ctx.getTelemetryService().saveAndNotify(
+                ctx.getTenantId(),
+                msg.getOriginator(),
+                eventIdentifier,
+                eventType,
+                new ArrayList<>(events),
+                new TelemetryNodeCallback(ctx, msg)
+        );
+    }
+
+    @Override
+    public void destroy() {
+    }
+
+}

+/**
+ * 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.rule.engine.yunteng.event;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.server.common.data.DataConstants;
+
+@Data
+public class TkMsgEventNodeConfiguration implements NodeConfiguration<TkMsgEventNodeConfiguration> {
+
+
+    private boolean useServerTs;
+    @Override
+    public TkMsgEventNodeConfiguration defaultConfiguration() {
+        TkMsgEventNodeConfiguration configuration = new TkMsgEventNodeConfiguration();
+        configuration.setUseServerTs(true);
+        return configuration;
+    }
+}