Browse Source

Serialziation via Mapstruct + Proto Draft

pull/8700/head
Andrii Shvaika 3 years ago
parent
commit
c15fa6f854
  1. 67
      common/cluster-api/src/main/proto/data.proto
  2. 4
      common/queue/pom.xml
  3. 34
      common/queue/src/main/java/org/thingsboard/server/queue/util/DeviceMapper.java
  4. 377
      common/queue/src/test/java/org/thingsboard/server/queue/util/DataDecodingEncodingServiceTest.java
  5. 11
      pom.xml

67
common/cluster-api/src/main/proto/data.proto

@ -0,0 +1,67 @@
/**
* Copyright © 2016-2023 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.
*/
syntax = "proto3";
package transport;
option java_package = "org.thingsboard.server.gen.data";
option java_multiple_files = true;
/**
* Data Serialization Structures;
*/
message DeviceProto {
string id = 1;
int64 createdTime = 3;
bytes additionalInfo = 4;
string tenantId = 5;
optional string customerId = 6;
optional string name = 7;
optional string type = 8;
optional string label = 9;
optional string deviceProfileId = 10;
bytes deviceDataBytes = 11;
optional string firmwareId = 12;
optional string softwareId = 13;
optional string externalId = 14;
}
message TbUUID {
}
message DeviceProto2 {
int64 idM = 1;
int64 idL = 2;
int64 createdTime = 3;
bytes additionalInfo = 4;
int64 tenantIdM = 5;
int64 tenantIdL = 6;
optional int64 customerIdM = 7;
optional int64 customerIdL = 8;
optional string name = 9;
optional string type = 10;
optional string label = 11;
optional int64 deviceProfileIdM = 12;
optional int64 deviceProfileIdL = 13;
bytes deviceDataBytes = 14;
optional int64 firmwareIdM = 15;
optional int64 firmwareIdL = 16;
optional int64 softwareIdM = 17;
optional int64 softwareIdL = 18;
optional int64 externalIdM = 19;
optional int64 externalIdL = 20;
}

4
common/queue/pom.xml

@ -120,6 +120,10 @@
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
</dependency>
<dependency>
<groupId>org.mapstruct</groupId>
<artifactId>mapstruct</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

34
common/queue/src/main/java/org/thingsboard/server/queue/util/DeviceMapper.java

@ -0,0 +1,34 @@
package org.thingsboard.server.queue.util;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.protobuf.ByteString;
import org.mapstruct.CollectionMappingStrategy;
import org.mapstruct.Mapper;
import org.mapstruct.NullValueCheckStrategy;
import org.mapstruct.factory.Mappers;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.id.UUIDBased;
import org.thingsboard.server.gen.data.DeviceProto;
import java.nio.charset.StandardCharsets;
@Mapper(collectionMappingStrategy = CollectionMappingStrategy.ADDER_PREFERRED,
nullValueCheckStrategy = NullValueCheckStrategy.ALWAYS)
public interface DeviceMapper {
DeviceMapper INSTANCE = Mappers.getMapper(DeviceMapper.class);
DeviceProto map(Device device);
default String map(UUIDBased id) {
return id.getId().toString();
}
default ByteString map(JsonNode additionalInfo) {
return ByteString.copyFrom(JacksonUtil.toString(additionalInfo), StandardCharsets.UTF_8);
}
// Device map(DeviceProto proto);
}

377
common/queue/src/test/java/org/thingsboard/server/queue/util/DataDecodingEncodingServiceTest.java

@ -1,293 +1,186 @@
package org.thingsboard.server.queue.util;
import lombok.SneakyThrows;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.util.SerializationUtils;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.FSTUtils;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UUIDBased;
import org.thingsboard.server.gen.data.DeviceProto;
import org.thingsboard.server.gen.data.DeviceProto2;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
public class DataDecodingEncodingServiceTest {
private static final int _1M = 1000 * 1000;
private static String DEVICE_PROFILE_STR = "{\n" +
" \"name\": \"thermostat\",\n" +
" \"description\": \"Thermostat device profile\",\n" +
" \"image\": null,\n" +
" \"type\": \"DEFAULT\",\n" +
" \"transportType\": \"DEFAULT\",\n" +
" \"provisionType\": \"DISABLED\",\n" +
" \"defaultRuleChainId\": {\n" +
" \"entityType\": \"RULE_CHAIN\",\n" +
" \"id\": \"eaaa47b0-e827-11ed-9f13-7f5975adc62f\"\n" +
" },\n" +
" \"defaultDashboardId\": null,\n" +
" \"defaultQueueName\": null,\n" +
" \"profileData\": {\n" +
" \"configuration\": {\n" +
" \"type\": \"DEFAULT\"\n" +
" },\n" +
" \"transportConfiguration\": {\n" +
" \"type\": \"DEFAULT\"\n" +
" },\n" +
" \"provisionConfiguration\": {\n" +
" \"type\": \"DISABLED\",\n" +
" \"provisionDeviceSecret\": null\n" +
" },\n" +
" \"alarms\": [\n" +
" {\n" +
" \"id\": \"highTemperatureAlarmID\",\n" +
" \"alarmType\": \"High Temperature\",\n" +
" \"createRules\": {\n" +
" \"MAJOR\": {\n" +
" \"condition\": {\n" +
" \"condition\": [\n" +
" {\n" +
" \"key\": {\n" +
" \"type\": \"ATTRIBUTE\",\n" +
" \"key\": \"temperatureAlarmFlag\"\n" +
" },\n" +
" \"valueType\": \"BOOLEAN\",\n" +
" \"value\": null,\n" +
" \"predicate\": {\n" +
" \"type\": \"BOOLEAN\",\n" +
" \"operation\": \"EQUAL\",\n" +
" \"value\": {\n" +
" \"defaultValue\": true,\n" +
" \"userValue\": null,\n" +
" \"dynamicValue\": null\n" +
" }\n" +
" }\n" +
" },\n" +
" {\n" +
" \"key\": {\n" +
" \"type\": \"TIME_SERIES\",\n" +
" \"key\": \"temperature\"\n" +
" },\n" +
" \"valueType\": \"NUMERIC\",\n" +
" \"value\": null,\n" +
" \"predicate\": {\n" +
" \"type\": \"NUMERIC\",\n" +
" \"operation\": \"GREATER\",\n" +
" \"value\": {\n" +
" \"defaultValue\": 25,\n" +
" \"userValue\": null,\n" +
" \"dynamicValue\": {\n" +
" \"sourceType\": \"CURRENT_DEVICE\",\n" +
" \"sourceAttribute\": \"temperatureAlarmThreshold\",\n" +
" \"inherit\": false\n" +
" }\n" +
" }\n" +
" }\n" +
" }\n" +
" ],\n" +
" \"spec\": {\n" +
" \"type\": \"SIMPLE\"\n" +
" }\n" +
" },\n" +
" \"schedule\": null,\n" +
" \"alarmDetails\": \"Current temperature = ${temperature}\",\n" +
" \"dashboardId\": null\n" +
" }\n" +
" },\n" +
" \"clearRule\": {\n" +
" \"condition\": {\n" +
" \"condition\": [\n" +
" {\n" +
" \"key\": {\n" +
" \"type\": \"TIME_SERIES\",\n" +
" \"key\": \"temperature\"\n" +
" },\n" +
" \"valueType\": \"NUMERIC\",\n" +
" \"value\": null,\n" +
" \"predicate\": {\n" +
" \"type\": \"NUMERIC\",\n" +
" \"operation\": \"LESS_OR_EQUAL\",\n" +
" \"value\": {\n" +
" \"defaultValue\": 25,\n" +
" \"userValue\": null,\n" +
" \"dynamicValue\": {\n" +
" \"sourceType\": \"CURRENT_DEVICE\",\n" +
" \"sourceAttribute\": \"temperatureAlarmThreshold\",\n" +
" \"inherit\": false\n" +
" }\n" +
" }\n" +
" }\n" +
" }\n" +
" ],\n" +
" \"spec\": {\n" +
" \"type\": \"SIMPLE\"\n" +
" }\n" +
" },\n" +
" \"schedule\": null,\n" +
" \"alarmDetails\": \"Current temperature = ${temperature}\",\n" +
" \"dashboardId\": null\n" +
" },\n" +
" \"propagate\": false,\n" +
" \"propagateToOwner\": false,\n" +
" \"propagateToTenant\": false,\n" +
" \"propagateRelationTypes\": null\n" +
" },\n" +
" {\n" +
" \"id\": \"lowHumidityAlarmID\",\n" +
" \"alarmType\": \"Low Humidity\",\n" +
" \"createRules\": {\n" +
" \"MINOR\": {\n" +
" \"condition\": {\n" +
" \"condition\": [\n" +
" {\n" +
" \"key\": {\n" +
" \"type\": \"ATTRIBUTE\",\n" +
" \"key\": \"humidityAlarmFlag\"\n" +
" },\n" +
" \"valueType\": \"BOOLEAN\",\n" +
" \"value\": null,\n" +
" \"predicate\": {\n" +
" \"type\": \"BOOLEAN\",\n" +
" \"operation\": \"EQUAL\",\n" +
" \"value\": {\n" +
" \"defaultValue\": true,\n" +
" \"userValue\": null,\n" +
" \"dynamicValue\": null\n" +
" }\n" +
" }\n" +
" },\n" +
" {\n" +
" \"key\": {\n" +
" \"type\": \"TIME_SERIES\",\n" +
" \"key\": \"humidity\"\n" +
" },\n" +
" \"valueType\": \"NUMERIC\",\n" +
" \"value\": null,\n" +
" \"predicate\": {\n" +
" \"type\": \"NUMERIC\",\n" +
" \"operation\": \"LESS\",\n" +
" \"value\": {\n" +
" \"defaultValue\": 60,\n" +
" \"userValue\": null,\n" +
" \"dynamicValue\": {\n" +
" \"sourceType\": \"CURRENT_DEVICE\",\n" +
" \"sourceAttribute\": \"humidityAlarmThreshold\",\n" +
" \"inherit\": false\n" +
" }\n" +
" }\n" +
" }\n" +
" }\n" +
" ],\n" +
" \"spec\": {\n" +
" \"type\": \"SIMPLE\"\n" +
" }\n" +
" },\n" +
" \"schedule\": null,\n" +
" \"alarmDetails\": \"Current humidity = ${humidity}\",\n" +
" \"dashboardId\": null\n" +
" }\n" +
" },\n" +
" \"clearRule\": {\n" +
" \"condition\": {\n" +
" \"condition\": [\n" +
" {\n" +
" \"key\": {\n" +
" \"type\": \"TIME_SERIES\",\n" +
" \"key\": \"humidity\"\n" +
" },\n" +
" \"valueType\": \"NUMERIC\",\n" +
" \"value\": null,\n" +
" \"predicate\": {\n" +
" \"type\": \"NUMERIC\",\n" +
" \"operation\": \"GREATER_OR_EQUAL\",\n" +
" \"value\": {\n" +
" \"defaultValue\": 60,\n" +
" \"userValue\": null,\n" +
" \"dynamicValue\": {\n" +
" \"sourceType\": \"CURRENT_DEVICE\",\n" +
" \"sourceAttribute\": \"humidityAlarmThreshold\",\n" +
" \"inherit\": false\n" +
" }\n" +
" }\n" +
" }\n" +
" }\n" +
" ],\n" +
" \"spec\": {\n" +
" \"type\": \"SIMPLE\"\n" +
" }\n" +
" },\n" +
" \"schedule\": null,\n" +
" \"alarmDetails\": \"Current humidity = ${humidity}\",\n" +
" \"dashboardId\": null\n" +
" },\n" +
" \"propagate\": false,\n" +
" \"propagateToOwner\": false,\n" +
" \"propagateToTenant\": false,\n" +
" \"propagateRelationTypes\": null\n" +
" }\n" +
" ]\n" +
" },\n" +
" \"provisionDeviceKey\": null,\n" +
" \"firmwareId\": null,\n" +
" \"softwareId\": null,\n" +
" \"defaultEdgeRuleChainId\": null,\n" +
" \"externalId\": null,\n" +
" \"default\": false\n" +
"}";
private static DeviceProfile dp;
private static Device d;
@BeforeAll
public static void before() {
dp = JacksonUtil.fromString(DEVICE_PROFILE_STR, DeviceProfile.class);
d = new Device();
d.setId(new DeviceId(UUID.randomUUID()));
d.setCreatedTime(System.currentTimeMillis());
d.setTenantId(new TenantId(UUID.randomUUID()));
d.setCustomerId(new CustomerId(UUID.randomUUID()));
d.setName(StringUtils.randomAlphabetic(10));
d.setLabel(StringUtils.randomAlphabetic(10));
d.setType(StringUtils.randomAlphabetic(10));
// d.setAdditionalInfo(JacksonUtil.newObjectNode().put("smth", StringUtils.randomAlphabetic(100)));
d.setFirmwareId(new OtaPackageId(UUID.randomUUID()));
d.setSoftwareId(new OtaPackageId(UUID.randomUUID()));
d.setExternalId(new DeviceId(UUID.randomUUID()));
}
@Ignore
@Test
void testDecode() {
void testSerialize() {
Assert.assertArrayEquals(toProto(d), toMapStruct(d));
// warmup
testDecode("FST", FSTUtils::encode);
testDecode("Jackson", DataDecodingEncodingServiceTest::serializeJackson);
testDecode("Default", SerializationUtils::serialize);
testSerialize("FST", FSTUtils::encode);
testSerialize("Proto", this::toProto);
testSerialize("Proto2", this::toProto2);
testSerialize("MapStruct", this::toMapStruct);
testSerialize("Jackson", DataDecodingEncodingServiceTest::serializeJackson);
// testSerialize("Default", SerializationUtils::serialize);
testDeserialize("FST", FSTUtils.encode(d), FSTUtils::decode);
testDeserialize("Proto", toProto(d), this::fromProto);
testDeserialize("Jackson", serializeJackson(d), data -> JacksonUtil.fromBytes(data, Device.class));
// testDeserialize("Default", serializeDefault(d), data -> (Device) SerializationUtils.deserialize(data));
}
private byte[] toMapStruct(Device device) {
return DeviceMapper.INSTANCE.map(device).toByteArray();
}
private byte[] toProto(Device device) {
var builder = DeviceProto.newBuilder();
builder.setId(device.getId().getId().toString());
builder.setCreatedTime(device.getCreatedTime());
builder.setTenantId(device.getTenantId().getId().toString());
toProto(device::getCustomerId, builder::setCustomerId);
builder.setName(device.getName());
builder.setLabel(device.getLabel());
builder.setType(device.getType());
toProto(device::getFirmwareId, builder::setFirmwareId);
toProto(device::getSoftwareId, builder::setSoftwareId);
toProto(device::getExternalId, builder::setExternalId);
return builder.build().toByteArray();
}
testEncode("FST", FSTUtils.encode(dp), FSTUtils::decode);
testEncode("Jackson", serializeJackson(dp), data -> JacksonUtil.fromBytes(data, DeviceProfile.class));
testEncode("Default", serializeDefault(dp), data -> (DeviceProfile) SerializationUtils.deserialize(data));
private byte[] toProto2(Device device) {
var builder = DeviceProto2.newBuilder();
builder.setIdM(device.getId().getId().getMostSignificantBits());
builder.setIdL(device.getId().getId().getLeastSignificantBits());
builder.setCreatedTime(device.getCreatedTime());
builder.setTenantIdM(device.getTenantId().getId().getMostSignificantBits());
builder.setTenantIdL(device.getTenantId().getId().getLeastSignificantBits());
toProto(device::getCustomerId, builder::setCustomerIdM, builder::setCustomerIdL);
builder.setName(device.getName());
builder.setLabel(device.getLabel());
builder.setType(device.getType());
toProto(device::getFirmwareId, builder::setFirmwareIdM, builder::setFirmwareIdL);
toProto(device::getSoftwareId, builder::setSoftwareIdM, builder::setSoftwareIdL);
toProto(device::getExternalId, builder::setExternalIdM, builder::setExternalIdL);
return builder.build().toByteArray();
}
private void testDecode(String methodName, Function<DeviceProfile, byte[]> serializeFunction) {
@SneakyThrows
private Device fromProto(byte[] bytes) {
DeviceProto dp = DeviceProto.parseFrom(bytes);
Device d = new Device();
d.setId(new DeviceId(UUID.fromString(dp.getId())));
d.setCreatedTime(dp.getCreatedTime());
d.setTenantId(new TenantId(UUID.fromString(dp.getTenantId())));
if (dp.hasCustomerId()) {
d.setCustomerId(new CustomerId(UUID.fromString(dp.getCustomerId())));
}
if (dp.hasName()) {
d.setName(dp.getName());
}
if (dp.hasLabel()) {
d.setLabel(dp.getLabel());
}
if (dp.hasType()) {
d.setType(dp.getType());
}
if (dp.hasFirmwareId()) {
d.setFirmwareId(new OtaPackageId(UUID.fromString(dp.getFirmwareId())));
}
if (dp.hasSoftwareId()) {
d.setSoftwareId(new OtaPackageId(UUID.fromString(dp.getSoftwareId())));
}
if (dp.hasExternalId()) {
d.setExternalId(new DeviceId(UUID.fromString(dp.getExternalId())));
}
return d;
}
private void toProto(Supplier<UUIDBased> extractId, Consumer<String> setId) {
UUIDBased id = extractId.get();
if (id != null) {
setId.accept(id.getId().toString());
}
}
private void toProto(Supplier<UUIDBased> extractId, Consumer<Long> setIdM, Consumer<Long> setIdL) {
UUIDBased id = extractId.get();
if (id != null) {
var uuid = id.getId();
setIdM.accept(uuid.getMostSignificantBits());
setIdL.accept(uuid.getLeastSignificantBits());
}
}
private void testSerialize(String methodName, Function<Device, byte[]> serializeFunction) {
// WARMUP;
for (int i = 0; i < 1000; i++) {
dp.setName(Integer.toString(i));
serializeFunction.apply(dp);
for (int i = 0; i < _1M; i++) {
d.setName(Integer.toString(i));
serializeFunction.apply(d);
}
long startTs = System.nanoTime();
for (int i = 0; i < _1M; i++) {
dp.setName(Integer.toString(i));
serializeFunction.apply(dp);
d.setName(Integer.toString(i));
serializeFunction.apply(d);
}
long elapsedTime = System.nanoTime() - startTs;
System.out.println("S: " + methodName + " SIZE: " + serializeFunction.apply(dp).length + " TIME: " + elapsedTime + "ns or " + (elapsedTime / 1000000) + "ms");
System.out.println("S: " + methodName + " SIZE: " + serializeFunction.apply(d).length + " TIME: " + (elapsedTime / 1000000) + "ms");
}
private void testEncode(String methodName, byte[] data, Function<byte[], DeviceProfile> deserializeFunction) {
private void testDeserialize(String methodName, byte[] data, Function<byte[], Device> deserializeFunction) {
// WARMUP;
for (int i = 0; i < 1000; i++) {
dp.setName(Integer.toString(i));
d.setName(Integer.toString(i));
deserializeFunction.apply(data);
}
long startTs = System.nanoTime();
for (int i = 0; i < _1M; i++) {
dp.setName(Integer.toString(i));
d.setName(Integer.toString(i));
deserializeFunction.apply(data);
}
long elapsedTime = System.nanoTime() - startTs;
System.out.println("D: " +methodName + " TIME: " + elapsedTime + "ns or " + (elapsedTime / 1000000) + "ms");
System.out.println("D: " + methodName + " TIME: " + (elapsedTime / 1000000) + "ms");
}
static <T> byte[] serializeDefault(T object) {

11
pom.xml

@ -80,6 +80,7 @@
<grpc.version>1.42.1</grpc.version>
<tbel.version>1.0.6</tbel.version>
<lombok.version>1.18.18</lombok.version>
<mapstruct.version>1.5.5.Final</mapstruct.version>
<paho.client.version>1.2.4</paho.client.version>
<paho.mqttv5.client.version>1.2.5</paho.mqttv5.client.version>
<netty.version>4.1.91.Final</netty.version>
@ -619,6 +620,11 @@
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
</path>
<path>
<groupId>org.mapstruct</groupId>
<artifactId>mapstruct-processor</artifactId>
<version>${mapstruct.version}</version>
</path>
</annotationProcessorPaths>
</configuration>
</plugin>
@ -1700,6 +1706,11 @@
<version>${dbunit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mapstruct</groupId>
<artifactId>mapstruct</artifactId>
<version>${mapstruct.version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>

Loading…
Cancel
Save