Merge pull request #10416 from AndriiLandiak/fix/kv-proto-util
Fix KvProtoUtils order for matching KeyValueType and DataType
This commit is contained in:
Andrew Shvayka
GitHub
commit
eab88dd500
@ -21,6 +21,7 @@ import com.google.common.util.concurrent.FutureCallback;
|
||||
import com.google.common.util.concurrent.Futures;
|
||||
import com.google.common.util.concurrent.ListenableFuture;
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
import jakarta.annotation.Nullable;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.collections.CollectionUtils;
|
||||
import org.thingsboard.common.util.JacksonUtil;
|
||||
@ -68,7 +69,6 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
|
||||
import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
|
||||
import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeoutMsg;
|
||||
import org.thingsboard.server.common.util.KvProtoUtil;
|
||||
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.DeviceSessionsCacheEntry;
|
||||
@ -97,7 +97,6 @@ import org.thingsboard.server.service.rpc.RpcSubmitStrategy;
|
||||
import org.thingsboard.server.service.state.DefaultDeviceStateService;
|
||||
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
|
||||
|
||||
import jakarta.annotation.Nullable;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
|
||||
@ -18,6 +18,8 @@ package org.thingsboard.server.service.queue;
|
||||
import com.google.common.util.concurrent.ListenableFuture;
|
||||
import com.google.common.util.concurrent.ListeningExecutorService;
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import lombok.Getter;
|
||||
import lombok.Setter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@ -50,9 +52,9 @@ import org.thingsboard.server.common.msg.queue.TbCallback;
|
||||
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
|
||||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequestActorMsg;
|
||||
import org.thingsboard.server.common.stats.StatsFactory;
|
||||
import org.thingsboard.server.common.util.KvProtoUtil;
|
||||
import org.thingsboard.server.common.util.ProtoUtils;
|
||||
import org.thingsboard.server.dao.resource.ImageCacheKey;
|
||||
import org.thingsboard.server.common.util.KvProtoUtil;
|
||||
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto;
|
||||
@ -101,8 +103,6 @@ import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWra
|
||||
import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate;
|
||||
import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate;
|
||||
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
@ -583,7 +583,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
|
||||
subscriptionManagerService.onTimeSeriesUpdate(
|
||||
toTenantId(tenantIdMSB, tenantIdLSB),
|
||||
TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
|
||||
KvProtoUtil.toTsKvEntityList(proto.getDataList()), callback);
|
||||
KvProtoUtil.fromTsKvProtoList(proto.getDataList()), callback);
|
||||
} else if (msg.hasAttrUpdate()) {
|
||||
TbAttributeUpdateProto proto = msg.getAttrUpdate();
|
||||
subscriptionManagerService.onAttributesUpdate(
|
||||
|
||||
@ -23,19 +23,9 @@ import org.thingsboard.server.common.data.id.EntityIdFactory;
|
||||
import org.thingsboard.server.common.data.id.TenantId;
|
||||
import org.thingsboard.server.common.data.id.UserId;
|
||||
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
|
||||
import org.thingsboard.server.common.data.kv.DataType;
|
||||
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.kv.TsKvEntry;
|
||||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmDeleteProto;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmUpdateProto;
|
||||
@ -46,30 +36,24 @@ import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesDeletePr
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
|
||||
import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate;
|
||||
import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate;
|
||||
import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate;
|
||||
import org.thingsboard.server.service.ws.telemetry.sub.AlarmSubscriptionUpdate;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.TreeMap;
|
||||
import java.util.UUID;
|
||||
|
||||
import static org.thingsboard.server.common.util.KvProtoUtil.fromTsValueProtoList;
|
||||
import static org.thingsboard.server.common.util.KvProtoUtil.toTsKvProtoBuilder;
|
||||
import static org.thingsboard.server.common.util.KvProtoUtil.toTsValueProto;
|
||||
|
||||
public class TbSubscriptionUtils {
|
||||
|
||||
private static final DataType[] dataTypeByProtoNumber;
|
||||
|
||||
static {
|
||||
int arraySize = Arrays.stream(DataType.values()).mapToInt(DataType::getProtoNumber).max().orElse(0);
|
||||
dataTypeByProtoNumber = new DataType[arraySize + 1];
|
||||
Arrays.stream(DataType.values()).forEach(dataType -> dataTypeByProtoNumber[dataType.getProtoNumber()] = dataType);
|
||||
}
|
||||
|
||||
public static ToCoreMsg toSubEventProto(String serviceId, TbEntitySubEvent event) {
|
||||
SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
|
||||
var builder = TbEntitySubEventProto.newBuilder()
|
||||
@ -189,7 +173,7 @@ public class TbSubscriptionUtils {
|
||||
builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits());
|
||||
builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
|
||||
builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
|
||||
ts.forEach(v -> builder.addData(toKeyValueProto(v.getTs(), v).build()));
|
||||
ts.forEach(v -> builder.addData(toTsKvProtoBuilder(v.getTs(), v).build()));
|
||||
SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
|
||||
msgBuilder.setTsUpdate(builder);
|
||||
return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
|
||||
@ -216,7 +200,7 @@ public class TbSubscriptionUtils {
|
||||
builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
|
||||
builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
|
||||
builder.setScope(scope);
|
||||
attributes.forEach(v -> builder.addData(toKeyValueProto(v.getLastUpdateTs(), v).build()));
|
||||
attributes.forEach(v -> builder.addData(toTsKvProtoBuilder(v.getLastUpdateTs(), v).build()));
|
||||
|
||||
SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
|
||||
msgBuilder.setAttrUpdate(builder);
|
||||
@ -239,121 +223,10 @@ public class TbSubscriptionUtils {
|
||||
return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
|
||||
}
|
||||
|
||||
private static TsKvProto.Builder toKeyValueProto(long ts, KvEntry attr) {
|
||||
KeyValueProto.Builder dataBuilder = KeyValueProto.newBuilder();
|
||||
dataBuilder.setKey(attr.getKey());
|
||||
dataBuilder.setType(toProto(attr.getDataType()));
|
||||
switch (attr.getDataType()) {
|
||||
case BOOLEAN:
|
||||
attr.getBooleanValue().ifPresent(dataBuilder::setBoolV);
|
||||
break;
|
||||
case LONG:
|
||||
attr.getLongValue().ifPresent(dataBuilder::setLongV);
|
||||
break;
|
||||
case DOUBLE:
|
||||
attr.getDoubleValue().ifPresent(dataBuilder::setDoubleV);
|
||||
break;
|
||||
case JSON:
|
||||
attr.getJsonValue().ifPresent(dataBuilder::setJsonV);
|
||||
break;
|
||||
case STRING:
|
||||
attr.getStrValue().ifPresent(dataBuilder::setStringV);
|
||||
break;
|
||||
}
|
||||
return TsKvProto.newBuilder().setTs(ts).setKv(dataBuilder);
|
||||
}
|
||||
|
||||
private static TransportProtos.TsValueProto toTsValueProto(long ts, KvEntry attr) {
|
||||
TransportProtos.TsValueProto.Builder dataBuilder = TransportProtos.TsValueProto.newBuilder();
|
||||
dataBuilder.setTs(ts);
|
||||
dataBuilder.setType(toProto(attr.getDataType()));
|
||||
switch (attr.getDataType()) {
|
||||
case BOOLEAN:
|
||||
attr.getBooleanValue().ifPresent(dataBuilder::setBoolV);
|
||||
break;
|
||||
case LONG:
|
||||
attr.getLongValue().ifPresent(dataBuilder::setLongV);
|
||||
break;
|
||||
case DOUBLE:
|
||||
attr.getDoubleValue().ifPresent(dataBuilder::setDoubleV);
|
||||
break;
|
||||
case JSON:
|
||||
attr.getJsonValue().ifPresent(dataBuilder::setJsonV);
|
||||
break;
|
||||
case STRING:
|
||||
attr.getStrValue().ifPresent(dataBuilder::setStringV);
|
||||
break;
|
||||
}
|
||||
return dataBuilder.build();
|
||||
}
|
||||
|
||||
|
||||
public static EntityId toEntityId(String entityType, long entityIdMSB, long entityIdLSB) {
|
||||
return EntityIdFactory.getByTypeAndUuid(entityType, new UUID(entityIdMSB, entityIdLSB));
|
||||
}
|
||||
|
||||
public static List<TsKvEntry> toTsKvEntityList(List<TsKvProto> dataList) {
|
||||
List<TsKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), getKvEntry(proto.getKv()))));
|
||||
return result;
|
||||
}
|
||||
|
||||
public static List<AttributeKvEntry> toAttributeKvList(List<TsKvProto> dataList) {
|
||||
List<AttributeKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BaseAttributeKvEntry(getKvEntry(proto.getKv()), proto.getTs())));
|
||||
return result;
|
||||
}
|
||||
|
||||
private static KvEntry getKvEntry(KeyValueProto proto) {
|
||||
KvEntry entry = null;
|
||||
switch (fromProto(proto.getType())) {
|
||||
case BOOLEAN:
|
||||
entry = new BooleanDataEntry(proto.getKey(), proto.getBoolV());
|
||||
break;
|
||||
case LONG:
|
||||
entry = new LongDataEntry(proto.getKey(), proto.getLongV());
|
||||
break;
|
||||
case DOUBLE:
|
||||
entry = new DoubleDataEntry(proto.getKey(), proto.getDoubleV());
|
||||
break;
|
||||
case STRING:
|
||||
entry = new StringDataEntry(proto.getKey(), proto.getStringV());
|
||||
break;
|
||||
case JSON:
|
||||
entry = new JsonDataEntry(proto.getKey(), proto.getJsonV());
|
||||
break;
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
public static List<TsKvEntry> toTsKvEntityList(String key, List<TransportProtos.TsValueProto> dataList) {
|
||||
List<TsKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), getKvEntry(key, proto))));
|
||||
return result;
|
||||
}
|
||||
|
||||
private static KvEntry getKvEntry(String key, TransportProtos.TsValueProto proto) {
|
||||
KvEntry entry = null;
|
||||
switch (fromProto(proto.getType())) {
|
||||
case BOOLEAN:
|
||||
entry = new BooleanDataEntry(key, proto.getBoolV());
|
||||
break;
|
||||
case LONG:
|
||||
entry = new LongDataEntry(key, proto.getLongV());
|
||||
break;
|
||||
case DOUBLE:
|
||||
entry = new DoubleDataEntry(key, proto.getDoubleV());
|
||||
break;
|
||||
case STRING:
|
||||
entry = new StringDataEntry(key, proto.getStringV());
|
||||
break;
|
||||
case JSON:
|
||||
entry = new JsonDataEntry(key, proto.getJsonV());
|
||||
break;
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
public static ToCoreMsg toAlarmUpdateProto(TenantId tenantId, EntityId entityId, AlarmInfo alarm) {
|
||||
TbAlarmUpdateProto.Builder builder = TbAlarmUpdateProto.newBuilder();
|
||||
builder.setEntityType(entityId.getEntityType().name());
|
||||
@ -411,7 +284,7 @@ public class TbSubscriptionUtils {
|
||||
public static List<TsKvEntry> fromProto(TransportProtos.TbSubUpdateProto proto) {
|
||||
List<TsKvEntry> result = new ArrayList<>();
|
||||
for (var p : proto.getDataList()) {
|
||||
result.addAll(toTsKvEntityList(p.getKey(), p.getTsValueList()));
|
||||
result.addAll(fromTsValueProtoList(p.getKey(), p.getTsValueList()));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
@ -453,12 +326,4 @@ public class TbSubscriptionUtils {
|
||||
return ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg(result).build();
|
||||
}
|
||||
|
||||
public static TransportProtos.KeyValueType toProto(DataType dataType) {
|
||||
return TransportProtos.KeyValueType.forNumber(dataType.getProtoNumber());
|
||||
}
|
||||
|
||||
public static DataType fromProto(TransportProtos.KeyValueType keyValueType) {
|
||||
return dataTypeByProtoNumber[keyValueType.getNumber()];
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -1,33 +0,0 @@
|
||||
/**
|
||||
* Copyright © 2016-2024 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.utils;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.thingsboard.server.common.data.kv.DataType;
|
||||
import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
public class TbSubscriptionUtilsTest {
|
||||
|
||||
@Test
|
||||
public void protoDataTypeSerialization() {
|
||||
for (DataType dataType : DataType.values()) {
|
||||
assertThat(TbSubscriptionUtils.fromProto(TbSubscriptionUtils.toProto(dataType))).as(dataType.name()).isEqualTo(dataType);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -29,11 +29,26 @@ import org.thingsboard.server.common.data.kv.TsKvEntry;
|
||||
import org.thingsboard.server.gen.transport.TransportProtos;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
public class KvProtoUtil {
|
||||
|
||||
private static final DataType[] dataTypeByProtoNumber;
|
||||
|
||||
static {
|
||||
int arraySize = Arrays.stream(DataType.values()).mapToInt(DataType::getProtoNumber).max().orElse(0);
|
||||
dataTypeByProtoNumber = new DataType[arraySize + 1];
|
||||
Arrays.stream(DataType.values()).forEach(dataType -> dataTypeByProtoNumber[dataType.getProtoNumber()] = dataType);
|
||||
}
|
||||
|
||||
public static List<AttributeKvEntry> toAttributeKvList(List<TransportProtos.TsKvProto> dataList) {
|
||||
List<AttributeKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BaseAttributeKvEntry(fromTsKvProto(proto.getKv()), proto.getTs())));
|
||||
return result;
|
||||
}
|
||||
|
||||
public static List<TransportProtos.TsKvProto> attrToTsKvProtos(List<AttributeKvEntry> result) {
|
||||
List<TransportProtos.TsKvProto> clientAttributes;
|
||||
if (result == null || result.isEmpty()) {
|
||||
@ -47,8 +62,7 @@ public class KvProtoUtil {
|
||||
return clientAttributes;
|
||||
}
|
||||
|
||||
|
||||
public static List<TransportProtos.TsKvProto> tsToTsKvProtos(List<TsKvEntry> result) {
|
||||
public static List<TransportProtos.TsKvProto> toTsKvProtoList(List<TsKvEntry> result) {
|
||||
List<TransportProtos.TsKvProto> ts;
|
||||
if (result == null || result.isEmpty()) {
|
||||
ts = Collections.emptyList();
|
||||
@ -61,95 +75,85 @@ public class KvProtoUtil {
|
||||
return ts;
|
||||
}
|
||||
|
||||
public static TransportProtos.TsKvProto toTsKvProto(long ts, KvEntry kvEntry) {
|
||||
return TransportProtos.TsKvProto.newBuilder().setTs(ts)
|
||||
.setKv(KvProtoUtil.toKeyValueProto(kvEntry)).build();
|
||||
public static List<TsKvEntry> fromTsKvProtoList(List<TransportProtos.TsKvProto> dataList) {
|
||||
List<TsKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), fromTsKvProto(proto.getKv()))));
|
||||
return result;
|
||||
}
|
||||
|
||||
public static TransportProtos.KeyValueProto toKeyValueProto(KvEntry kvEntry) {
|
||||
public static TransportProtos.TsKvProto toTsKvProto(long ts, KvEntry kvEntry) {
|
||||
return TransportProtos.TsKvProto.newBuilder().setTs(ts)
|
||||
.setKv(KvProtoUtil.toKeyValueTypeProto(kvEntry)).build();
|
||||
}
|
||||
|
||||
public static TsKvEntry fromTsKvProto(TransportProtos.TsKvProto proto) {
|
||||
return new BasicTsKvEntry(proto.getTs(), fromTsKvProto(proto.getKv()));
|
||||
}
|
||||
|
||||
public static TransportProtos.KeyValueProto toKeyValueTypeProto(KvEntry kvEntry) {
|
||||
TransportProtos.KeyValueProto.Builder builder = TransportProtos.KeyValueProto.newBuilder();
|
||||
builder.setKey(kvEntry.getKey());
|
||||
builder.setType(toKeyValueTypeProto(kvEntry.getDataType()));
|
||||
switch (kvEntry.getDataType()) {
|
||||
case BOOLEAN:
|
||||
builder.setType(TransportProtos.KeyValueType.BOOLEAN_V);
|
||||
builder.setBoolV(kvEntry.getBooleanValue().get());
|
||||
break;
|
||||
case DOUBLE:
|
||||
builder.setType(TransportProtos.KeyValueType.DOUBLE_V);
|
||||
builder.setDoubleV(kvEntry.getDoubleValue().get());
|
||||
break;
|
||||
case LONG:
|
||||
builder.setType(TransportProtos.KeyValueType.LONG_V);
|
||||
builder.setLongV(kvEntry.getLongValue().get());
|
||||
break;
|
||||
case STRING:
|
||||
builder.setType(TransportProtos.KeyValueType.STRING_V);
|
||||
builder.setStringV(kvEntry.getStrValue().get());
|
||||
break;
|
||||
case JSON:
|
||||
builder.setType(TransportProtos.KeyValueType.JSON_V);
|
||||
builder.setJsonV(kvEntry.getJsonValue().get());
|
||||
break;
|
||||
case BOOLEAN -> kvEntry.getBooleanValue().ifPresent(builder::setBoolV);
|
||||
case LONG -> kvEntry.getLongValue().ifPresent(builder::setLongV);
|
||||
case DOUBLE -> kvEntry.getDoubleValue().ifPresent(builder::setDoubleV);
|
||||
case JSON -> kvEntry.getJsonValue().ifPresent(builder::setJsonV);
|
||||
case STRING -> kvEntry.getStrValue().ifPresent(builder::setStringV);
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
public static TransportProtos.TsKvProto.Builder toKeyValueProto(long ts, KvEntry attr) {
|
||||
TransportProtos.KeyValueProto.Builder dataBuilder = TransportProtos.KeyValueProto.newBuilder();
|
||||
dataBuilder.setKey(attr.getKey());
|
||||
dataBuilder.setType(TransportProtos.KeyValueType.forNumber(attr.getDataType().ordinal()));
|
||||
switch (attr.getDataType()) {
|
||||
case BOOLEAN:
|
||||
attr.getBooleanValue().ifPresent(dataBuilder::setBoolV);
|
||||
break;
|
||||
case LONG:
|
||||
attr.getLongValue().ifPresent(dataBuilder::setLongV);
|
||||
break;
|
||||
case DOUBLE:
|
||||
attr.getDoubleValue().ifPresent(dataBuilder::setDoubleV);
|
||||
break;
|
||||
case JSON:
|
||||
attr.getJsonValue().ifPresent(dataBuilder::setJsonV);
|
||||
break;
|
||||
case STRING:
|
||||
attr.getStrValue().ifPresent(dataBuilder::setStringV);
|
||||
break;
|
||||
}
|
||||
return TransportProtos.TsKvProto.newBuilder().setTs(ts).setKv(dataBuilder);
|
||||
public static KvEntry fromTsKvProto(TransportProtos.KeyValueProto proto) {
|
||||
return switch (fromKeyValueTypeProto(proto.getType())) {
|
||||
case BOOLEAN -> new BooleanDataEntry(proto.getKey(), proto.getBoolV());
|
||||
case LONG -> new LongDataEntry(proto.getKey(), proto.getLongV());
|
||||
case DOUBLE -> new DoubleDataEntry(proto.getKey(), proto.getDoubleV());
|
||||
case STRING -> new StringDataEntry(proto.getKey(), proto.getStringV());
|
||||
case JSON -> new JsonDataEntry(proto.getKey(), proto.getJsonV());
|
||||
};
|
||||
}
|
||||
|
||||
public static List<TsKvEntry> toTsKvEntityList(List<TransportProtos.TsKvProto> dataList) {
|
||||
public static TransportProtos.TsKvProto.Builder toTsKvProtoBuilder(long ts, KvEntry kvEntry) {
|
||||
return TransportProtos.TsKvProto.newBuilder().setTs(ts).setKv(KvProtoUtil.toKeyValueTypeProto(kvEntry));
|
||||
}
|
||||
|
||||
public static List<TsKvEntry> fromTsValueProtoList(String key, List<TransportProtos.TsValueProto> dataList) {
|
||||
List<TsKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), getKvEntry(proto.getKv()))));
|
||||
dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), fromTsValueProto(key, proto))));
|
||||
return result;
|
||||
}
|
||||
|
||||
public static List<AttributeKvEntry> toAttributeKvList(List<TransportProtos.TsKvProto> dataList) {
|
||||
List<AttributeKvEntry> result = new ArrayList<>(dataList.size());
|
||||
dataList.forEach(proto -> result.add(new BaseAttributeKvEntry(getKvEntry(proto.getKv()), proto.getTs())));
|
||||
return result;
|
||||
}
|
||||
|
||||
private static KvEntry getKvEntry(TransportProtos.KeyValueProto proto) {
|
||||
KvEntry entry = null;
|
||||
DataType type = DataType.values()[proto.getType().getNumber()];
|
||||
switch (type) {
|
||||
case BOOLEAN:
|
||||
entry = new BooleanDataEntry(proto.getKey(), proto.getBoolV());
|
||||
break;
|
||||
case LONG:
|
||||
entry = new LongDataEntry(proto.getKey(), proto.getLongV());
|
||||
break;
|
||||
case DOUBLE:
|
||||
entry = new DoubleDataEntry(proto.getKey(), proto.getDoubleV());
|
||||
break;
|
||||
case STRING:
|
||||
entry = new StringDataEntry(proto.getKey(), proto.getStringV());
|
||||
break;
|
||||
case JSON:
|
||||
entry = new JsonDataEntry(proto.getKey(), proto.getJsonV());
|
||||
break;
|
||||
public static TransportProtos.TsValueProto toTsValueProto(long ts, KvEntry attr) {
|
||||
TransportProtos.TsValueProto.Builder dataBuilder = TransportProtos.TsValueProto.newBuilder();
|
||||
dataBuilder.setTs(ts);
|
||||
dataBuilder.setType(toKeyValueTypeProto(attr.getDataType()));
|
||||
switch (attr.getDataType()) {
|
||||
case BOOLEAN -> attr.getBooleanValue().ifPresent(dataBuilder::setBoolV);
|
||||
case LONG -> attr.getLongValue().ifPresent(dataBuilder::setLongV);
|
||||
case DOUBLE -> attr.getDoubleValue().ifPresent(dataBuilder::setDoubleV);
|
||||
case JSON -> attr.getJsonValue().ifPresent(dataBuilder::setJsonV);
|
||||
case STRING -> attr.getStrValue().ifPresent(dataBuilder::setStringV);
|
||||
}
|
||||
return entry;
|
||||
return dataBuilder.build();
|
||||
}
|
||||
|
||||
public static KvEntry fromTsValueProto(String key, TransportProtos.TsValueProto proto) {
|
||||
return switch (fromKeyValueTypeProto(proto.getType())) {
|
||||
case BOOLEAN -> new BooleanDataEntry(key, proto.getBoolV());
|
||||
case LONG -> new LongDataEntry(key, proto.getLongV());
|
||||
case DOUBLE -> new DoubleDataEntry(key, proto.getDoubleV());
|
||||
case STRING -> new StringDataEntry(key, proto.getStringV());
|
||||
case JSON -> new JsonDataEntry(key, proto.getJsonV());
|
||||
};
|
||||
}
|
||||
|
||||
public static TransportProtos.KeyValueType toKeyValueTypeProto(DataType dataType) {
|
||||
return TransportProtos.KeyValueType.forNumber(dataType.getProtoNumber());
|
||||
}
|
||||
|
||||
public static DataType fromKeyValueTypeProto(TransportProtos.KeyValueType keyValueType) {
|
||||
return dataTypeByProtoNumber[keyValueType.getNumber()];
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -0,0 +1,118 @@
|
||||
/**
|
||||
* Copyright © 2016-2024 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.util;
|
||||
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.EnumSource;
|
||||
import org.junit.jupiter.params.provider.MethodSource;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import org.thingsboard.server.common.data.kv.AggTsKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
|
||||
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
|
||||
import org.thingsboard.server.common.data.kv.DataType;
|
||||
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.kv.TsKvEntry;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class KvProtoUtilTest {
|
||||
|
||||
private static final long TS = System.currentTimeMillis();
|
||||
|
||||
private static Stream<KvEntry> kvEntryData() {
|
||||
String key = "key";
|
||||
return Stream.of(
|
||||
new BooleanDataEntry(key, true),
|
||||
new LongDataEntry(key, 23L),
|
||||
new DoubleDataEntry(key, 23.0),
|
||||
new StringDataEntry(key, "stringValue"),
|
||||
new JsonDataEntry(key, "jsonValue")
|
||||
);
|
||||
}
|
||||
|
||||
private static Stream<KvEntry> basicTsKvEntryData() {
|
||||
return kvEntryData().map(kvEntry -> new BasicTsKvEntry(TS, kvEntry));
|
||||
}
|
||||
|
||||
private static Stream<List<BaseAttributeKvEntry>> attributeKvEntryData() {
|
||||
return Stream.of(kvEntryData().map(kvEntry -> new BaseAttributeKvEntry(TS, kvEntry)).toList());
|
||||
}
|
||||
|
||||
private static List<TsKvEntry> createTsKvEntryList(boolean withAggregation) {
|
||||
return kvEntryData().map(kvEntry -> {
|
||||
if (withAggregation) {
|
||||
return new AggTsKvEntry(TS, kvEntry, 0);
|
||||
} else {
|
||||
return new BasicTsKvEntry(TS, kvEntry);
|
||||
}
|
||||
}).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@EnumSource(DataType.class)
|
||||
void protoDataTypeSerialization(DataType dataType) {
|
||||
assertThat(KvProtoUtil.fromKeyValueTypeProto(KvProtoUtil.toKeyValueTypeProto(dataType)))
|
||||
.as(dataType.name()).isEqualTo(dataType);
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@MethodSource("kvEntryData")
|
||||
void protoKeyValueProtoSerialization(KvEntry kvEntry) {
|
||||
assertThat(KvProtoUtil.fromTsKvProto(KvProtoUtil.toKeyValueTypeProto(kvEntry)))
|
||||
.as("deserialized").isEqualTo(kvEntry);
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@MethodSource("basicTsKvEntryData")
|
||||
void protoTsKvEntrySerialization(KvEntry kvEntry) {
|
||||
assertThat(KvProtoUtil.fromTsKvProto(KvProtoUtil.toTsKvProto(TS, kvEntry)))
|
||||
.as("deserialized").isEqualTo(kvEntry);
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@MethodSource("kvEntryData")
|
||||
void protoTsValueSerialization(KvEntry kvEntry) {
|
||||
assertThat(KvProtoUtil.fromTsValueProto(kvEntry.getKey(), KvProtoUtil.toTsValueProto(TS, kvEntry)))
|
||||
.as("deserialized").isEqualTo(kvEntry);
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(booleans = {true, false})
|
||||
void protoListTsKvEntrySerialization(boolean withAggregation) {
|
||||
List<TsKvEntry> tsKvEntries = createTsKvEntryList(withAggregation);
|
||||
assertThat(KvProtoUtil.fromTsKvProtoList(KvProtoUtil.toTsKvProtoList(tsKvEntries)))
|
||||
.as("deserialized").isEqualTo(tsKvEntries);
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@MethodSource("attributeKvEntryData")
|
||||
void protoListAttributeKvSerialization(List<AttributeKvEntry> attributeKvEntries) {
|
||||
assertThat(KvProtoUtil.toAttributeKvList(KvProtoUtil.attrToTsKvProtos(attributeKvEntries)))
|
||||
.as("deserialized")
|
||||
.isEqualTo(attributeKvEntries);
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user