diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index d6b5da4ddf..51205e97c9 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -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; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index fd08b9da4d..cdb3f19601 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -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 dataTypeByProtoNumber[dataType.getProtoNumber()] = dataType); - } +public class TbSubscriptionUtils { public static ToCoreMsg toSubEventProto(String serviceId, TbEntitySubEvent event) { SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.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 toTsKvEntityList(List dataList) { - List result = new ArrayList<>(dataList.size()); - dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), getKvEntry(proto.getKv())))); - return result; - } - - public static List toAttributeKvList(List dataList) { - List 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 toTsKvEntityList(String key, List dataList) { - List 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 fromProto(TransportProtos.TbSubUpdateProto proto) { List 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()]; - } - } diff --git a/application/src/test/java/org/thingsboard/server/utils/TbSubscriptionUtilsTest.java b/application/src/test/java/org/thingsboard/server/utils/TbSubscriptionUtilsTest.java deleted file mode 100644 index a3f1f0f0fa..0000000000 --- a/application/src/test/java/org/thingsboard/server/utils/TbSubscriptionUtilsTest.java +++ /dev/null @@ -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); - } - } - -} diff --git a/common/proto/src/main/java/org/thingsboard/server/common/util/KvProtoUtil.java b/common/proto/src/main/java/org/thingsboard/server/common/util/KvProtoUtil.java index 504ffe368d..74674e1e45 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/util/KvProtoUtil.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/util/KvProtoUtil.java @@ -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 toAttributeKvList(List dataList) { + List result = new ArrayList<>(dataList.size()); + dataList.forEach(proto -> result.add(new BaseAttributeKvEntry(fromTsKvProto(proto.getKv()), proto.getTs()))); + return result; + } + public static List attrToTsKvProtos(List result) { List clientAttributes; if (result == null || result.isEmpty()) { @@ -47,8 +62,7 @@ public class KvProtoUtil { return clientAttributes; } - - public static List tsToTsKvProtos(List result) { + public static List toTsKvProtoList(List result) { List ts; if (result == null || result.isEmpty()) { ts = Collections.emptyList(); @@ -61,95 +75,85 @@ public class KvProtoUtil { return ts; } + public static List fromTsKvProtoList(List dataList) { + List result = new ArrayList<>(dataList.size()); + dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), fromTsKvProto(proto.getKv())))); + return result; + } + public static TransportProtos.TsKvProto toTsKvProto(long ts, KvEntry kvEntry) { return TransportProtos.TsKvProto.newBuilder().setTs(ts) - .setKv(KvProtoUtil.toKeyValueProto(kvEntry)).build(); + .setKv(KvProtoUtil.toKeyValueTypeProto(kvEntry)).build(); + } + + public static TsKvEntry fromTsKvProto(TransportProtos.TsKvProto proto) { + return new BasicTsKvEntry(proto.getTs(), fromTsKvProto(proto.getKv())); } - public static TransportProtos.KeyValueProto toKeyValueProto(KvEntry kvEntry) { + 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 toTsKvEntityList(List dataList) { - List result = new ArrayList<>(dataList.size()); - dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), getKvEntry(proto.getKv())))); - return result; + public static TransportProtos.TsKvProto.Builder toTsKvProtoBuilder(long ts, KvEntry kvEntry) { + return TransportProtos.TsKvProto.newBuilder().setTs(ts).setKv(KvProtoUtil.toKeyValueTypeProto(kvEntry)); } - public static List toAttributeKvList(List dataList) { - List result = new ArrayList<>(dataList.size()); - dataList.forEach(proto -> result.add(new BaseAttributeKvEntry(getKvEntry(proto.getKv()), proto.getTs()))); + public static List fromTsValueProtoList(String key, List dataList) { + List result = new ArrayList<>(dataList.size()); + dataList.forEach(proto -> result.add(new BasicTsKvEntry(proto.getTs(), fromTsValueProto(key, proto)))); 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()]; + } + } diff --git a/common/proto/src/test/java/org/thingsboard/server/common/util/KvProtoUtilTest.java b/common/proto/src/test/java/org/thingsboard/server/common/util/KvProtoUtilTest.java new file mode 100644 index 0000000000..72d71c7c93 --- /dev/null +++ b/common/proto/src/test/java/org/thingsboard/server/common/util/KvProtoUtilTest.java @@ -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 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 basicTsKvEntryData() { + return kvEntryData().map(kvEntry -> new BasicTsKvEntry(TS, kvEntry)); + } + + private static Stream> attributeKvEntryData() { + return Stream.of(kvEntryData().map(kvEntry -> new BaseAttributeKvEntry(TS, kvEntry)).toList()); + } + + private static List 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 tsKvEntries = createTsKvEntryList(withAggregation); + assertThat(KvProtoUtil.fromTsKvProtoList(KvProtoUtil.toTsKvProtoList(tsKvEntries))) + .as("deserialized").isEqualTo(tsKvEntries); + } + + @ParameterizedTest + @MethodSource("attributeKvEntryData") + void protoListAttributeKvSerialization(List attributeKvEntries) { + assertThat(KvProtoUtil.toAttributeKvList(KvProtoUtil.attrToTsKvProtos(attributeKvEntries))) + .as("deserialized") + .isEqualTo(attributeKvEntries); + } + +}