diff --git a/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java index 89ef53b19d..74523852a9 100644 --- a/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java @@ -15,35 +15,11 @@ */ package org.thingsboard.server.transport; -import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; -import org.eclipse.paho.client.mqttv3.MqttAsyncClient; -import org.eclipse.paho.client.mqttv3.MqttConnectOptions; -import org.eclipse.paho.client.mqttv3.MqttException; -import org.eclipse.paho.client.mqttv3.MqttMessage; -import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; -import org.junit.Assert; -import org.springframework.util.StringUtils; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; -import org.thingsboard.server.common.data.DeviceProfileProvisionType; -import org.thingsboard.server.common.data.DeviceProfileType; -import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.Tenant; -import org.thingsboard.server.common.data.TransportPayloadType; import org.thingsboard.server.common.data.User; -import org.thingsboard.server.common.data.device.profile.AllowCreateNewDevicesDeviceProfileProvisionConfiguration; -import org.thingsboard.server.common.data.device.profile.CheckPreProvisionedDevicesDeviceProfileProvisionConfiguration; -import org.thingsboard.server.common.data.device.profile.DefaultDeviceProfileConfiguration; -import org.thingsboard.server.common.data.device.profile.DeviceProfileData; -import org.thingsboard.server.common.data.device.profile.DeviceProfileProvisionConfiguration; -import org.thingsboard.server.common.data.device.profile.DisabledDeviceProfileProvisionConfiguration; -import org.thingsboard.server.common.data.device.profile.JsonTransportPayloadConfiguration; -import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration; -import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration; -import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; -import org.thingsboard.server.common.data.security.Authority; -import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.gen.transport.TransportProtos; @@ -51,8 +27,6 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @Slf4j @@ -105,6 +79,22 @@ public abstract class AbstractTransportIntegrationTest extends AbstractControlle " }\n" + "}"; + protected static final String DEVICE_RPC_RESPONSE_PROTO_SCHEMA = "syntax =\"proto3\";\n" + + "package rpc;\n" + + "\n" + + "message RpcResponseMsg {\n" + + " string payload = 1;\n" + + "}"; + + protected static final String DEVICE_RPC_REQUEST_PROTO_SCHEMA = "syntax =\"proto3\";\n" + + "package rpc;\n" + + "\n" + + "message RpcRequestMsg {\n" + + " string method = 1;\n" + + " int32 requestId = 2;\n" + + " string params = 3;\n" + + "}"; + protected Tenant savedTenant; protected User tenantAdmin; diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java index ff64d70423..83023e21b9 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/AbstractCoapIntegrationTest.java @@ -53,7 +53,7 @@ import static org.junit.Assert.assertNotNull; public abstract class AbstractCoapIntegrationTest extends AbstractTransportIntegrationTest { protected void processBeforeTest(String deviceName, CoapDeviceType coapDeviceType, TransportPayloadType payloadType) throws Exception { - this.processBeforeTest(deviceName, coapDeviceType, payloadType, null, null, DeviceProfileProvisionType.DISABLED, null, null); + this.processBeforeTest(deviceName, coapDeviceType, payloadType, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); } protected void processBeforeTest(String deviceName, @@ -61,8 +61,11 @@ public abstract class AbstractCoapIntegrationTest extends AbstractTransportInteg TransportPayloadType payloadType, String telemetryProtoSchema, String attributesProtoSchema, - DeviceProfileProvisionType provisionType, - String provisionKey, String provisionSecret + String rpcResponseProtoSchema, + String rpcRequestProtoSchema, + String provisionKey, + String provisionSecret, + DeviceProfileProvisionType provisionType ) throws Exception { loginSysAdmin(); @@ -85,7 +88,7 @@ public abstract class AbstractCoapIntegrationTest extends AbstractTransportInteg device.setType("default"); if (coapDeviceType != null) { - DeviceProfile coapDeviceProfile = createCoapDeviceProfile(payloadType, coapDeviceType, attributesProtoSchema, provisionType, provisionKey, provisionSecret, telemetryProtoSchema); + DeviceProfile coapDeviceProfile = createCoapDeviceProfile(payloadType, coapDeviceType, provisionSecret, provisionType, provisionKey, attributesProtoSchema, telemetryProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema); deviceProfile = doPost("/api/deviceProfile", coapDeviceProfile, DeviceProfile.class); device.setType(deviceProfile.getName()); device.setDeviceProfileId(deviceProfile.getId()); @@ -103,8 +106,9 @@ public abstract class AbstractCoapIntegrationTest extends AbstractTransportInteg } protected DeviceProfile createCoapDeviceProfile(TransportPayloadType transportPayloadType, CoapDeviceType coapDeviceType, - String attributesProtoSchema, DeviceProfileProvisionType provisionType, - String provisionKey, String provisionSecret, String telemetryProtoSchema) { + String provisionSecret, DeviceProfileProvisionType provisionType, + String provisionKey, String attributesProtoSchema, + String telemetryProtoSchema, String rpcResponseProtoSchema, String rpcRequestProtoSchema) { DeviceProfile deviceProfile = new DeviceProfile(); deviceProfile.setName(transportPayloadType.name()); deviceProfile.setType(DeviceProfileType.DEFAULT); @@ -127,8 +131,16 @@ public abstract class AbstractCoapIntegrationTest extends AbstractTransportInteg if (StringUtils.isEmpty(attributesProtoSchema)) { attributesProtoSchema = DEVICE_ATTRIBUTES_PROTO_SCHEMA; } + if (StringUtils.isEmpty(rpcResponseProtoSchema)) { + rpcResponseProtoSchema = DEVICE_RPC_RESPONSE_PROTO_SCHEMA; + } + if (StringUtils.isEmpty(rpcRequestProtoSchema)) { + rpcRequestProtoSchema = DEVICE_RPC_REQUEST_PROTO_SCHEMA; + } protoTransportPayloadConfiguration.setDeviceTelemetryProtoSchema(telemetryProtoSchema); protoTransportPayloadConfiguration.setDeviceAttributesProtoSchema(attributesProtoSchema); + protoTransportPayloadConfiguration.setDeviceRpcResponseProtoSchema(rpcResponseProtoSchema); + protoTransportPayloadConfiguration.setDeviceRpcRequestProtoSchema(rpcRequestProtoSchema); transportPayloadTypeConfiguration = protoTransportPayloadConfiguration; } else { transportPayloadTypeConfiguration = new JsonTransportPayloadConfiguration(); diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/AbstractCoapAttributesRequestProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/AbstractCoapAttributesRequestProtoIntegrationTest.java index 4fbe103430..5bef44434e 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/AbstractCoapAttributesRequestProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/AbstractCoapAttributesRequestProtoIntegrationTest.java @@ -79,7 +79,7 @@ public abstract class AbstractCoapAttributesRequestProtoIntegrationTest extends @Test public void testRequestAttributesValuesFromTheServer() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", CoapDeviceType.DEFAULT, - TransportPayloadType.PROTOBUF, null, ATTRIBUTES_SCHEMA_STR, DeviceProfileProvisionType.DISABLED, null, null); + TransportPayloadType.PROTOBUF, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED); processTestRequestAttributesValuesFromTheServer(); } diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionJsonDeviceTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionJsonDeviceTest.java index b4a55ae271..caecaea8ec 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionJsonDeviceTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionJsonDeviceTest.java @@ -88,7 +88,7 @@ public abstract class AbstractCoapProvisionJsonDeviceTest extends AbstractCoapIn private void processTestProvisioningDisabledDevice() throws Exception { - super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); byte[] result = createCoapClientAndPublish().getPayload(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString()); @@ -97,7 +97,7 @@ public abstract class AbstractCoapProvisionJsonDeviceTest extends AbstractCoapIn private void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { - super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); byte[] result = createCoapClientAndPublish().getPayload(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -113,7 +113,7 @@ public abstract class AbstractCoapProvisionJsonDeviceTest extends AbstractCoapIn private void processTestProvisioningCreateNewDeviceWithAccessToken() throws Exception { - super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); String requestCredentials = ",\"credentialsType\": \"ACCESS_TOKEN\",\"token\": \"test_token\""; byte[] result = createCoapClientAndPublish(requestCredentials).getPayload(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -132,7 +132,7 @@ public abstract class AbstractCoapProvisionJsonDeviceTest extends AbstractCoapIn private void processTestProvisioningCreateNewDeviceWithCert() throws Exception { - super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); String requestCredentials = ",\"credentialsType\": \"X509_CERTIFICATE\",\"hash\": \"testHash\""; byte[] result = createCoapClientAndPublish(requestCredentials).getPayload(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -156,7 +156,7 @@ public abstract class AbstractCoapProvisionJsonDeviceTest extends AbstractCoapIn } private void processTestProvisioningCheckPreProvisionedDevice() throws Exception { - super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); byte[] result = createCoapClientAndPublish().getPayload(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -167,7 +167,7 @@ public abstract class AbstractCoapProvisionJsonDeviceTest extends AbstractCoapIn } private void processTestProvisioningWithBadKeyDevice() throws Exception { - super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKeyOrig", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.JSON, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); byte[] result = createCoapClientAndPublish().getPayload(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString()); diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionProtoDeviceTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionProtoDeviceTest.java index 9f1d8f1f18..682d9c7e7b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionProtoDeviceTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/provision/AbstractCoapProvisionProtoDeviceTest.java @@ -92,14 +92,14 @@ public abstract class AbstractCoapProvisionProtoDeviceTest extends AbstractCoapI private void processTestProvisioningDisabledDevice() throws Exception { - super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); ProvisionDeviceResponseMsg result = ProvisionDeviceResponseMsg.parseFrom(createCoapClientAndPublish().getPayload()); Assert.assertNotNull(result); Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), result.getStatus().toString()); } private void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { - super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createCoapClientAndPublish().getPayload()); Device createdDevice = deviceService.findDeviceByTenantIdAndName(savedTenant.getTenantId(), "Test Provision device"); @@ -113,7 +113,7 @@ public abstract class AbstractCoapProvisionProtoDeviceTest extends AbstractCoapI } private void processTestProvisioningCreateNewDeviceWithAccessToken() throws Exception { - super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateDeviceTokenRequestMsg(ValidateDeviceTokenRequestMsg.newBuilder().setToken("test_token").build()).build(); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createCoapClientAndPublish(createTestsProvisionMessage(CredentialsType.ACCESS_TOKEN, requestCredentials)).getPayload()); @@ -131,7 +131,7 @@ public abstract class AbstractCoapProvisionProtoDeviceTest extends AbstractCoapI } private void processTestProvisioningCreateNewDeviceWithCert() throws Exception { - super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateDeviceX509CertRequestMsg(ValidateDeviceX509CertRequestMsg.newBuilder().setHash("testHash").build()).build(); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createCoapClientAndPublish(createTestsProvisionMessage(CredentialsType.X509_CERTIFICATE, requestCredentials)).getPayload()); @@ -155,7 +155,7 @@ public abstract class AbstractCoapProvisionProtoDeviceTest extends AbstractCoapI } private void processTestProvisioningCheckPreProvisionedDevice() throws Exception { - super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createCoapClientAndPublish().getPayload()); DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(savedTenant.getTenantId(), savedDevice.getId()); @@ -165,7 +165,7 @@ public abstract class AbstractCoapProvisionProtoDeviceTest extends AbstractCoapI } private void processTestProvisioningWithBadKeyDevice() throws Exception { - super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKeyOrig", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createCoapClientAndPublish().getPayload()); Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), response.getStatus().toString()); } diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcIntegrationTest.java index a36660267e..d5d56db63b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcIntegrationTest.java @@ -124,7 +124,7 @@ public abstract class AbstractCoapServerSideRpcIntegrationTest extends AbstractC return COAP_BASE_URL + token + "/" + FeatureType.RPC.name().toLowerCase() + "/" + requestId; } - private class TestCoapCallback implements CoapHandler { + protected class TestCoapCallback implements CoapHandler { private final CoapClient client; private final CountDownLatch latch; @@ -136,7 +136,7 @@ public abstract class AbstractCoapServerSideRpcIntegrationTest extends AbstractC private Integer observe; - private TestCoapCallback(CoapClient client, CountDownLatch latch, boolean isOneWayRpc) { + TestCoapCallback(CoapClient client, CountDownLatch latch, boolean isOneWayRpc) { this.client = client; this.latch = latch; this.isOneWayRpc = isOneWayRpc; @@ -144,7 +144,7 @@ public abstract class AbstractCoapServerSideRpcIntegrationTest extends AbstractC @Override public void onLoad(CoapResponse response) { - log.warn("coap response: {}, {}", response, response.getCode()); + log.warn("coap response: {}, {}", response.getResponseText(), response.getCode()); assertNotNull(response.getPayload()); assertEquals(response.getCode(), CoAP.ResponseCode.CONTENT); observe = response.getOptions().getObserve(); diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcJsonIntegrationTest.java index 5387d3c7bf..cecdb1fae0 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcJsonIntegrationTest.java @@ -36,12 +36,12 @@ public abstract class AbstractCoapServerSideRpcJsonIntegrationTest extends Abstr } @Test - public void testServerMqttOneWayRpc() throws Exception { + public void testServerCoapOneWayRpc() throws Exception { processOneWayRpcTest(); } @Test - public void testServerMqttTwoWayRpc() throws Exception { + public void testServerCoapTwoWayRpc() throws Exception { processTwoWayRpcTest(); } diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcProtoIntegrationTest.java index edbdf69848..706a560024 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcProtoIntegrationTest.java @@ -15,26 +15,62 @@ */ package org.thingsboard.server.transport.coap.rpc; +import com.github.os72.protobuf.dynamic.DynamicSchema; +import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.InvalidProtocolBufferException; +import com.squareup.wire.schema.internal.parser.ProtoFileElement; import lombok.extern.slf4j.Slf4j; import org.eclipse.californium.core.CoapClient; import org.eclipse.californium.core.CoapHandler; +import org.eclipse.californium.core.CoapObserveRelation; import org.eclipse.californium.core.CoapResponse; +import org.eclipse.californium.core.coap.CoAP; import org.eclipse.californium.core.coap.MediaTypeRegistry; +import org.eclipse.californium.core.coap.Request; import org.junit.After; import org.junit.Before; import org.junit.Test; import org.thingsboard.server.common.data.CoapDeviceType; +import org.thingsboard.server.common.data.DeviceProfileProvisionType; import org.thingsboard.server.common.data.TransportPayloadType; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.common.data.device.profile.CoapDeviceProfileTransportConfiguration; +import org.thingsboard.server.common.data.device.profile.CoapDeviceTypeConfiguration; +import org.thingsboard.server.common.data.device.profile.DefaultCoapDeviceTypeConfiguration; +import org.thingsboard.server.common.data.device.profile.DeviceProfileTransportConfiguration; +import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration; +import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; +import org.thingsboard.server.common.msg.session.FeatureType; +import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @Slf4j public abstract class AbstractCoapServerSideRpcProtoIntegrationTest extends AbstractCoapServerSideRpcIntegrationTest { + private static final String RPC_REQUEST_PROTO_SCHEMA = "syntax =\"proto3\";\n" + + "package rpc;\n" + + "\n" + + "message RpcRequestMsg {\n" + + " string method = 1;\n" + + " int32 requestId = 2;\n" + + " Params params = 3;\n" + + "\n" + + " message Params {\n" + + " string pin = 1;\n" + + " int32 value = 2;\n" + + " }\n" + + "}"; + @Before public void beforeTest() throws Exception { - processBeforeTest("RPC test device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF); + processBeforeTest("RPC test device", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED); } @After @@ -43,33 +79,91 @@ public abstract class AbstractCoapServerSideRpcProtoIntegrationTest extends Abst } @Test - public void testServerMqttOneWayRpc() throws Exception { + public void testServerCoapOneWayRpc() throws Exception { processOneWayRpcTest(); } @Test - public void testServerMqttTwoWayRpc() throws Exception { + public void testServerCoapTwoWayRpc() throws Exception { processTwoWayRpcTest(); } + protected void processTwoWayRpcTest() throws Exception { + CoapClient client = getCoapClient(FeatureType.RPC); + client.useCONs(); + + CountDownLatch latch = new CountDownLatch(1); + TestCoapCallback testCoapCallback = new TestCoapCallback(client, latch, false); + + Request request = Request.newGet().setObserve(); + request.setType(CoAP.Type.CON); + CoapObserveRelation observeRelation = client.observe(request, testCoapCallback); + + String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}"; + String deviceId = savedDevice.getId().getId().toString(); + + String expected = "{\"payload\":\"{\\\"value1\\\":\\\"A\\\",\\\"value2\\\":\\\"B\\\"}\"}"; + + String result = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk()); + latch.await(3, TimeUnit.SECONDS); + + assertEquals(expected, result); + assertEquals(0, testCoapCallback.getObserve().intValue()); + observeRelation.proactiveCancel(); + assertTrue(observeRelation.isCanceled()); + } + @Override protected void processOnLoadResponse(CoapResponse response, CoapClient client, Integer observe, CountDownLatch latch) { client.setURI(getRpcResponseFeatureTokenUrl(accessToken, observe)); - TransportProtos.ToDeviceRpcResponseMsg toDeviceRpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() - .setPayload(DEVICE_RESPONSE) - .setRequestId(observe) - .build(); - client.post(new CoapHandler() { - @Override - public void onLoad(CoapResponse response) { - log.warn("Command Response Ack: {}, {}", response.getCode(), response.getResponseText()); - latch.countDown(); - } + ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = getProtoTransportPayloadConfiguration(); + ProtoFileElement rpcRequestProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(RPC_REQUEST_PROTO_SCHEMA); + DynamicSchema rpcRequestProtoSchema = protoTransportPayloadConfiguration.getDynamicSchema(rpcRequestProtoSchemaFile, ProtoTransportPayloadConfiguration.RPC_REQUEST_PROTO_SCHEMA); - @Override - public void onError() { - log.warn("Command Response Ack Error, No connect"); + byte[] requestPayload = response.getPayload(); + DynamicMessage.Builder rpcRequestMsg = rpcRequestProtoSchema.newMessageBuilder("RpcRequestMsg"); + Descriptors.Descriptor rpcRequestMsgDescriptor = rpcRequestMsg.getDescriptorForType(); + assertNotNull(rpcRequestMsgDescriptor); + try { + DynamicMessage dynamicMessage = DynamicMessage.parseFrom(rpcRequestMsgDescriptor, requestPayload); + List fields = rpcRequestMsgDescriptor.getFields(); + for (Descriptors.FieldDescriptor fieldDescriptor: fields) { + assertTrue(dynamicMessage.hasField(fieldDescriptor)); } - }, toDeviceRpcResponseMsg.toByteArray(), MediaTypeRegistry.APPLICATION_JSON); + ProtoFileElement rpcResponseProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(DEVICE_RPC_RESPONSE_PROTO_SCHEMA); + DynamicSchema rpcResponseProtoSchema = protoTransportPayloadConfiguration.getDynamicSchema(rpcResponseProtoSchemaFile, ProtoTransportPayloadConfiguration.RPC_RESPONSE_PROTO_SCHEMA); + DynamicMessage.Builder rpcResponseBuilder = rpcResponseProtoSchema.newMessageBuilder("RpcResponseMsg"); + Descriptors.Descriptor rpcResponseMsgDescriptor = rpcResponseBuilder.getDescriptorForType(); + assertNotNull(rpcResponseMsgDescriptor); + DynamicMessage rpcResponseMsg = rpcResponseBuilder + .setField(rpcResponseMsgDescriptor.findFieldByName("payload"), DEVICE_RESPONSE) + .build(); + client.post(new CoapHandler() { + @Override + public void onLoad(CoapResponse response) { + log.warn("Command Response Ack: {}, {}", response.getCode(), response.getResponseText()); + latch.countDown(); + } + + @Override + public void onError() { + log.warn("Command Response Ack Error, No connect"); + } + }, rpcResponseMsg.toByteArray(), MediaTypeRegistry.APPLICATION_JSON); + } catch (InvalidProtocolBufferException e) { + log.warn("Command Response Ack Error, Invalid response received: ", e); + } + } + + private ProtoTransportPayloadConfiguration getProtoTransportPayloadConfiguration() { + DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); + assertTrue(transportConfiguration instanceof CoapDeviceProfileTransportConfiguration); + CoapDeviceProfileTransportConfiguration coapDeviceProfileTransportConfiguration = (CoapDeviceProfileTransportConfiguration) transportConfiguration; + CoapDeviceTypeConfiguration coapDeviceTypeConfiguration = coapDeviceProfileTransportConfiguration.getCoapDeviceTypeConfiguration(); + assertTrue(coapDeviceTypeConfiguration instanceof DefaultCoapDeviceTypeConfiguration); + DefaultCoapDeviceTypeConfiguration defaultCoapDeviceTypeConfiguration = (DefaultCoapDeviceTypeConfiguration) coapDeviceTypeConfiguration; + TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = defaultCoapDeviceTypeConfiguration.getTransportPayloadTypeConfiguration(); + assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration); + return (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration; } } diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/telemetry/timeseries/AbstractCoapTimeseriesProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/telemetry/timeseries/AbstractCoapTimeseriesProtoIntegrationTest.java index 2ae686b211..0cc3db9cf1 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/telemetry/timeseries/AbstractCoapTimeseriesProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/telemetry/timeseries/AbstractCoapTimeseriesProtoIntegrationTest.java @@ -114,7 +114,7 @@ public abstract class AbstractCoapTimeseriesProtoIntegrationTest extends Abstrac " }\n" + " }\n" + "}"; - super.processBeforeTest("Test Post Telemetry device proto payload", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, schemaStr, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Post Telemetry device proto payload", CoapDeviceType.DEFAULT, TransportPayloadType.PROTOBUF, schemaStr, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); assertTrue(transportConfiguration instanceof CoapDeviceProfileTransportConfiguration); CoapDeviceProfileTransportConfiguration coapDeviceProfileTransportConfiguration = (CoapDeviceProfileTransportConfiguration) transportConfiguration; diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java index cd074b3197..0d11c9a773 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java @@ -62,7 +62,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg protected DeviceProfile deviceProfile; protected void processBeforeTest (String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic) throws Exception { - this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, DeviceProfileProvisionType.DISABLED, null, null); + this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); } protected void processBeforeTest(String deviceName, @@ -72,9 +72,12 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg String attributesTopic, String telemetryProtoSchema, String attributesProtoSchema, - DeviceProfileProvisionType provisionType, - String provisionKey, String provisionSecret - ) throws Exception { + String rpcResponseProtoSchema, + String rpcRequestProtoSchema, + String provisionKey, + String provisionSecret, + DeviceProfileProvisionType provisionType + ) throws Exception { loginSysAdmin(); Tenant tenant = new Tenant(); @@ -103,7 +106,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg gateway.setAdditionalInfo(additionalInfo); if (payloadType != null) { - DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic, telemetryProtoSchema, attributesProtoSchema, provisionType, provisionKey, provisionSecret); + DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic, telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema, provisionKey, provisionSecret, provisionType); deviceProfile = doPost("/api/deviceProfile", mqttDeviceProfile, DeviceProfile.class); device.setType(deviceProfile.getName()); device.setDeviceProfileId(deviceProfile.getId()); @@ -157,8 +160,9 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg protected DeviceProfile createMqttDeviceProfile(TransportPayloadType transportPayloadType, String telemetryTopic, String attributesTopic, String telemetryProtoSchema, String attributesProtoSchema, - DeviceProfileProvisionType provisionType, - String provisionKey, String provisionSecret) { + String rpcResponseProtoSchema, String rpcRequestProtoSchema, + String provisionKey, String provisionSecret, + DeviceProfileProvisionType provisionType) { DeviceProfile deviceProfile = new DeviceProfile(); deviceProfile.setName(transportPayloadType.name()); deviceProfile.setType(DeviceProfileType.DEFAULT); @@ -186,8 +190,16 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg if (StringUtils.isEmpty(attributesProtoSchema)) { attributesProtoSchema = DEVICE_ATTRIBUTES_PROTO_SCHEMA; } + if (StringUtils.isEmpty(rpcResponseProtoSchema)) { + rpcResponseProtoSchema = DEVICE_RPC_RESPONSE_PROTO_SCHEMA; + } + if (StringUtils.isEmpty(rpcRequestProtoSchema)) { + rpcRequestProtoSchema = DEVICE_RPC_REQUEST_PROTO_SCHEMA; + } protoTransportPayloadConfiguration.setDeviceTelemetryProtoSchema(telemetryProtoSchema); protoTransportPayloadConfiguration.setDeviceAttributesProtoSchema(attributesProtoSchema); + protoTransportPayloadConfiguration.setDeviceRpcResponseProtoSchema(rpcResponseProtoSchema); + protoTransportPayloadConfiguration.setDeviceRpcRequestProtoSchema(rpcRequestProtoSchema); transportPayloadTypeConfiguration = protoTransportPayloadConfiguration; } mqttDeviceProfileTransportConfiguration.setTransportPayloadTypeConfiguration(transportPayloadTypeConfiguration); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java index ff8b54c79c..6949a544de 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java @@ -83,7 +83,7 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends @Test public void testRequestAttributesValuesFromTheServer() throws Exception { super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto", - TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, DeviceProfileProvisionType.DISABLED, null, null); + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED); processTestRequestAttributesValuesFromTheServer(); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java index 106c801676..870b62b195 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java @@ -94,7 +94,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningDisabledDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString()); @@ -103,7 +103,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -119,7 +119,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithAccessToken() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); String requestCredentials = ",\"credentialsType\": \"ACCESS_TOKEN\",\"token\": \"test_token\""; byte[] result = createMqttClientAndPublish(requestCredentials).getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -138,7 +138,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithCert() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); String requestCredentials = ",\"credentialsType\": \"X509_CERTIFICATE\",\"hash\": \"testHash\""; byte[] result = createMqttClientAndPublish(requestCredentials).getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -163,7 +163,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn protected void processTestProvisioningCreateNewDeviceWithMqttBasic() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); String requestCredentials = ",\"credentialsType\": \"MQTT_BASIC\",\"clientId\": \"test_clientId\",\"username\": \"test_username\",\"password\": \"test_password\""; byte[] result = createMqttClientAndPublish(requestCredentials).getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -188,7 +188,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn } protected void processTestProvisioningCheckPreProvisionedDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); @@ -199,7 +199,7 @@ public abstract class AbstractMqttProvisionJsonDeviceTest extends AbstractMqttIn } protected void processTestProvisioningWithBadKeyDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKeyOrig", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); byte[] result = createMqttClientAndPublish().getPayloadBytes(); JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject(); Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString()); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java index 53fde9412b..97ebbfe4a7 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java @@ -101,14 +101,14 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI protected void processTestProvisioningDisabledDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); ProvisionDeviceResponseMsg result = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); Assert.assertNotNull(result); Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), result.getStatus().toString()); } protected void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); Device createdDevice = deviceService.findDeviceByTenantIdAndName(savedTenant.getTenantId(), "Test Provision device"); @@ -122,7 +122,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCreateNewDeviceWithAccessToken() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null,null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null,null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateDeviceTokenRequestMsg(ValidateDeviceTokenRequestMsg.newBuilder().setToken("test_token").build()).build(); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish(createTestsProvisionMessage(CredentialsType.ACCESS_TOKEN, requestCredentials)).getPayloadBytes()); @@ -140,7 +140,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCreateNewDeviceWithCert() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateDeviceX509CertRequestMsg(ValidateDeviceX509CertRequestMsg.newBuilder().setHash("testHash").build()).build(); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish(createTestsProvisionMessage(CredentialsType.X509_CERTIFICATE, requestCredentials)).getPayloadBytes()); @@ -164,7 +164,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCreateNewDeviceWithMqttBasic() throws Exception { - super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES); CredentialsDataProto requestCredentials = CredentialsDataProto.newBuilder().setValidateBasicMqttCredRequestMsg( ValidateBasicMqttCredRequestMsg.newBuilder() .setClientId("test_clientId") @@ -195,7 +195,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningCheckPreProvisionedDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKey", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(savedTenant.getTenantId(), savedDevice.getId()); @@ -205,7 +205,7 @@ public abstract class AbstractMqttProvisionProtoDeviceTest extends AbstractMqttI } protected void processTestProvisioningWithBadKeyDevice() throws Exception { - super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, "testProvisionKeyOrig", "testProvisionSecret"); + super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES); ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes()); Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), response.getStatus().toString()); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java index 90de461627..f24cb0c2bc 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java @@ -180,7 +180,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM return message; } - private class TestMqttCallback implements MqttCallback { + protected class TestMqttCallback implements MqttCallback { private final MqttAsyncClient client; private final CountDownLatch latch; diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java index 575dfdb052..eea800fb0e 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java @@ -15,25 +15,60 @@ */ package org.thingsboard.server.transport.mqtt.rpc; +import com.github.os72.protobuf.dynamic.DynamicSchema; +import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; import com.google.protobuf.InvalidProtocolBufferException; +import com.squareup.wire.schema.internal.parser.ProtoFileElement; import lombok.extern.slf4j.Slf4j; +import org.eclipse.californium.core.CoapHandler; +import org.eclipse.californium.core.CoapResponse; +import org.eclipse.californium.core.coap.MediaTypeRegistry; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.jetbrains.annotations.NotNull; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.thingsboard.server.common.data.DeviceProfileProvisionType; import org.thingsboard.server.common.data.TransportPayloadType; +import org.thingsboard.server.common.data.device.profile.DeviceProfileTransportConfiguration; +import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration; import org.thingsboard.server.common.data.device.profile.MqttTopics; +import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration; +import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; import org.thingsboard.server.gen.transport.TransportApiProtos; -import org.thingsboard.server.gen.transport.TransportProtos; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @Slf4j public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends AbstractMqttServerSideRpcIntegrationTest { + private static final String RPC_REQUEST_PROTO_SCHEMA = "syntax =\"proto3\";\n" + + "package rpc;\n" + + "\n" + + "message RpcRequestMsg {\n" + + " string method = 1;\n" + + " int32 requestId = 2;\n" + + " Params params = 3;\n" + + "\n" + + " message Params {\n" + + " string pin = 1;\n" + + " int32 value = 2;\n" + + " }\n" + + "}"; + @Before public void beforeTest() throws Exception { - processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null); + processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED); } @After @@ -83,14 +118,55 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst return builder.build(); } + protected void processTwoWayRpcTest() throws Exception { + MqttAsyncClient client = getMqttAsyncClient(accessToken); + client.subscribe(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC, 1); + + CountDownLatch latch = new CountDownLatch(1); + TestMqttCallback callback = new TestMqttCallback(client, latch); + client.setCallback(callback); + + Thread.sleep(1000); + + String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}"; + String deviceId = savedDevice.getId().getId().toString(); + + String result = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk()); + String expected = "{\"payload\":\"{\\\"value1\\\":\\\"A\\\",\\\"value2\\\":\\\"B\\\"}\"}"; + latch.await(3, TimeUnit.SECONDS); + Assert.assertEquals(expected, result); + } + protected MqttMessage processMessageArrived(String requestTopic, MqttMessage mqttMessage) throws MqttException, InvalidProtocolBufferException { MqttMessage message = new MqttMessage(); if (requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC)) { - TransportProtos.ToDeviceRpcResponseMsg toDeviceRpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() - .setPayload(DEVICE_RESPONSE) - .setRequestId(0) + ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = getProtoTransportPayloadConfiguration(); + ProtoFileElement rpcRequestProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(RPC_REQUEST_PROTO_SCHEMA); + DynamicSchema rpcRequestProtoSchema = protoTransportPayloadConfiguration.getDynamicSchema(rpcRequestProtoSchemaFile, ProtoTransportPayloadConfiguration.RPC_REQUEST_PROTO_SCHEMA); + + byte[] requestPayload = mqttMessage.getPayload(); + DynamicMessage.Builder rpcRequestMsg = rpcRequestProtoSchema.newMessageBuilder("RpcRequestMsg"); + Descriptors.Descriptor rpcRequestMsgDescriptor = rpcRequestMsg.getDescriptorForType(); + assertNotNull(rpcRequestMsgDescriptor); + try { + DynamicMessage dynamicMessage = DynamicMessage.parseFrom(rpcRequestMsgDescriptor, requestPayload); + List fields = rpcRequestMsgDescriptor.getFields(); + for (Descriptors.FieldDescriptor fieldDescriptor: fields) { + assertTrue(dynamicMessage.hasField(fieldDescriptor)); + } + ProtoFileElement transportProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(DEVICE_RPC_RESPONSE_PROTO_SCHEMA); + DynamicSchema rpcResponseProtoSchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchemaFile, ProtoTransportPayloadConfiguration.RPC_RESPONSE_PROTO_SCHEMA); + + DynamicMessage.Builder rpcResponseBuilder = rpcResponseProtoSchema.newMessageBuilder("RpcResponseMsg"); + Descriptors.Descriptor rpcResponseMsgDescriptor = rpcResponseBuilder.getDescriptorForType(); + assertNotNull(rpcResponseMsgDescriptor); + DynamicMessage rpcResponseMsg = rpcResponseBuilder + .setField(rpcResponseMsgDescriptor.findFieldByName("payload"), DEVICE_RESPONSE) .build(); - message.setPayload(toDeviceRpcResponseMsg.toByteArray()); + message.setPayload(rpcResponseMsg.toByteArray()); + } catch (InvalidProtocolBufferException e) { + log.warn("Command Response Ack Error, Invalid response received: ", e); + } } else { TransportApiProtos.GatewayDeviceRpcRequestMsg msg = TransportApiProtos.GatewayDeviceRpcRequestMsg.parseFrom(mqttMessage.getPayload()); String deviceName = msg.getDeviceName(); @@ -105,6 +181,14 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst return message; } + private ProtoTransportPayloadConfiguration getProtoTransportPayloadConfiguration() { + DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); + assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration); + MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration; + TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration(); + assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration); + return (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration; + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java index 93a004318d..6140b7d8ed 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java @@ -119,7 +119,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac " }\n" + " }\n" + "}"; - super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, schemaStr, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, schemaStr, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration); @@ -172,7 +172,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac @Test public void testPushMqttTelemetryGateway() throws Exception { - super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, null, null, null, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); TransportApiProtos.GatewayTelemetryMsg.Builder gatewayTelemetryMsgProtoBuilder = TransportApiProtos.GatewayTelemetryMsg.newBuilder(); List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); String deviceName1 = "Device A"; @@ -186,7 +186,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac @Test public void testGatewayConnect() throws Exception { - super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, null, null, DeviceProfileProvisionType.DISABLED, null, null); + super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED); String deviceName = "Device A"; TransportApiProtos.ConnectMsg connectMsgProto = getConnectProto(deviceName); MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/ProtoTransportPayloadConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/ProtoTransportPayloadConfiguration.java index 13982f7e0b..159fbced16 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/ProtoTransportPayloadConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/ProtoTransportPayloadConfiguration.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.common.data.device.profile; -import com.fasterxml.jackson.annotation.JsonIgnore; import com.github.os72.protobuf.dynamic.DynamicSchema; import com.github.os72.protobuf.dynamic.EnumDefinition; import com.github.os72.protobuf.dynamic.MessageDefinition; @@ -46,9 +45,13 @@ public class ProtoTransportPayloadConfiguration implements TransportPayloadTypeC public static final Location LOCATION = new Location("", "", -1, -1); public static final String ATTRIBUTES_PROTO_SCHEMA = "attributes proto schema"; public static final String TELEMETRY_PROTO_SCHEMA = "telemetry proto schema"; + public static final String RPC_RESPONSE_PROTO_SCHEMA = "rpc response proto schema"; + public static final String RPC_REQUEST_PROTO_SCHEMA = "rpc request proto schema"; private String deviceTelemetryProtoSchema; private String deviceAttributesProtoSchema; + private String deviceRpcRequestProtoSchema; + private String deviceRpcResponseProtoSchema; @Override public TransportPayloadType getTransportPayloadType() { @@ -63,13 +66,45 @@ public class ProtoTransportPayloadConfiguration implements TransportPayloadTypeC return getDescriptor(deviceAttributesProtoSchema, ATTRIBUTES_PROTO_SCHEMA); } + public Descriptors.Descriptor getRpcResponseDynamicMessageDescriptor(String deviceRpcResponseProtoSchema) { + return getDescriptor(deviceRpcResponseProtoSchema, RPC_RESPONSE_PROTO_SCHEMA); + } + + public DynamicMessage.Builder getRpcRequestDynamicMessageBuilder(String deviceRpcRequestProtoSchema) { + return getDynamicMessageBuilder(deviceRpcRequestProtoSchema, RPC_REQUEST_PROTO_SCHEMA); + } + + public String getDeviceRpcResponseProtoSchema() { + if (!isEmptyStr(deviceRpcResponseProtoSchema)) { + return deviceRpcResponseProtoSchema; + } else { + return "syntax =\"proto3\";\n" + + "package rpc;\n" + + "\n" + + "message RpcResponseMsg {\n" + + " string payload = 1;\n" + + "}"; + } + } + + public String getDeviceRpcRequestProtoSchema() { + if (!isEmptyStr(deviceRpcRequestProtoSchema)) { + return deviceRpcRequestProtoSchema; + } else { + return "syntax =\"proto3\";\n" + + "package rpc;\n" + + "\n" + + "message RpcRequestMsg {\n" + + " string method = 1;\n" + + " int32 requestId = 2;\n" + + " string params = 3;\n" + + "}"; + } + } + private Descriptors.Descriptor getDescriptor(String protoSchema, String schemaName) { try { - ProtoFileElement protoFileElement = getTransportProtoSchema(protoSchema); - DynamicSchema dynamicSchema = getDynamicSchema(protoFileElement, schemaName); - String lastMsgName = getMessageTypes(protoFileElement.getTypes()).stream() - .map(MessageElement::getName).reduce((previous, last) -> last).get(); - DynamicMessage.Builder builder = dynamicSchema.newMessageBuilder(lastMsgName); + DynamicMessage.Builder builder = getDynamicMessageBuilder(protoSchema, schemaName); return builder.getDescriptorForType(); } catch (Exception e) { log.warn("Failed to get Message Descriptor due to {}", e.getMessage()); @@ -77,6 +112,14 @@ public class ProtoTransportPayloadConfiguration implements TransportPayloadTypeC } } + public DynamicMessage.Builder getDynamicMessageBuilder(String protoSchema, String schemaName) { + ProtoFileElement protoFileElement = getTransportProtoSchema(protoSchema); + DynamicSchema dynamicSchema = getDynamicSchema(protoFileElement, schemaName); + String lastMsgName = getMessageTypes(protoFileElement.getTypes()).stream() + .map(MessageElement::getName).reduce((previous, last) -> last).get(); + return dynamicSchema.newMessageBuilder(lastMsgName); + } + public DynamicSchema getDynamicSchema(ProtoFileElement protoFileElement, String schemaName) { DynamicSchema.Builder schemaBuilder = DynamicSchema.newBuilder(); schemaBuilder.setName(schemaName); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java index 8d756631e8..dcbb62e0c5 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java @@ -17,6 +17,7 @@ package org.thingsboard.server.transport.coap; import com.google.gson.JsonParseException; import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.eclipse.californium.core.coap.CoAP; @@ -261,7 +262,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource { TransportProtos.SessionInfoProto currentAttrSession = tokenToSessionIdMap.get(getTokenFromRequest(request)); if (currentAttrSession == null) { attributeSubscriptions.add(sessionId); - registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor, getTokenFromRequest(request)); + registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor, + transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(), getTokenFromRequest(request)); transportService.process(sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), new CoapNoOpCallback(exchange)); } @@ -281,7 +283,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource { TransportProtos.SessionInfoProto currentRpcSession = tokenToSessionIdMap.get(getTokenFromRequest(request)); if (currentRpcSession == null) { rpcSubscriptions.add(sessionId); - registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor, getTokenFromRequest(request)); + registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor, + transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(), getTokenFromRequest(request)); transportService.process(sessionInfo, TransportProtos.SubscribeToRPCMsg.getDefaultInstance(), new CoapNoOpCallback(exchange)); @@ -303,17 +306,19 @@ public class CoapTransportResource extends AbstractCoapTransportResource { break; case TO_DEVICE_RPC_RESPONSE: transportService.process(sessionInfo, - coapTransportAdaptor.convertToDeviceRpcResponse(sessionId, request), + coapTransportAdaptor.convertToDeviceRpcResponse(sessionId, request, transportConfigurationContainer.getRpcResponseMsgDescriptor()), new CoapOkCallback(exchange, CoAP.ResponseCode.CREATED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); break; case TO_SERVER_RPC_REQUEST: - transportService.registerSyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor), timeout); + transportService.registerSyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor, + transportConfigurationContainer.getRpcRequestDynamicMessageBuilder()), timeout); transportService.process(sessionInfo, coapTransportAdaptor.convertToServerRpcRequest(sessionId, request), new CoapNoOpCallback(exchange)); break; case GET_ATTRIBUTES_REQUEST: - transportService.registerSyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor), timeout); + transportService.registerSyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor, + transportConfigurationContainer.getRpcRequestDynamicMessageBuilder()), timeout); transportService.process(sessionInfo, coapTransportAdaptor.convertToGetAttributes(sessionId, request), new CoapNoOpCallback(exchange)); @@ -330,14 +335,14 @@ public class CoapTransportResource extends AbstractCoapTransportResource { return tokenToSessionIdMap.remove(token); } - private void registerAsyncCoapSession(CoapExchange exchange, TransportProtos.SessionInfoProto sessionInfo, CoapTransportAdaptor coapTransportAdaptor, String token) { + private void registerAsyncCoapSession(CoapExchange exchange, TransportProtos.SessionInfoProto sessionInfo, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, String token) { tokenToSessionIdMap.putIfAbsent(token, sessionInfo); - transportService.registerAsyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor)); + transportService.registerAsyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder)); transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null); } - private CoapSessionListener getCoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor) { - return new CoapSessionListener(exchange, coapTransportAdaptor); + private CoapSessionListener getCoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) { + return new CoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder); } private String getTokenFromRequest(Request request) { @@ -423,10 +428,12 @@ public class CoapTransportResource extends AbstractCoapTransportResource { private final CoapExchange exchange; private final CoapTransportAdaptor coapTransportAdaptor; + private final DynamicMessage.Builder rpcRequestDynamicMessageBuilder; - CoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor) { + CoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) { this.exchange = exchange; this.coapTransportAdaptor = coapTransportAdaptor; + this.rpcRequestDynamicMessageBuilder = rpcRequestDynamicMessageBuilder; } @Override @@ -457,7 +464,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource { @Override public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg msg) { try { - exchange.respond(coapTransportAdaptor.convertToPublish(isConRequest(), msg)); + exchange.respond(coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder)); } catch (AdaptorException e) { log.trace("Failed to reply due to error", e); exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR); @@ -545,9 +552,14 @@ public class CoapTransportResource extends AbstractCoapTransportResource { (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration; String deviceTelemetryProtoSchema = protoTransportPayloadConfiguration.getDeviceTelemetryProtoSchema(); String deviceAttributesProtoSchema = protoTransportPayloadConfiguration.getDeviceAttributesProtoSchema(); + String deviceRpcRequestProtoSchema = protoTransportPayloadConfiguration.getDeviceRpcRequestProtoSchema(); + String deviceRpcResponseProtoSchema = protoTransportPayloadConfiguration.getDeviceRpcResponseProtoSchema(); return new TransportConfigurationContainer(false, protoTransportPayloadConfiguration.getTelemetryDynamicMessageDescriptor(deviceTelemetryProtoSchema), - protoTransportPayloadConfiguration.getAttributesDynamicMessageDescriptor(deviceAttributesProtoSchema)); + protoTransportPayloadConfiguration.getAttributesDynamicMessageDescriptor(deviceAttributesProtoSchema), + protoTransportPayloadConfiguration.getRpcResponseDynamicMessageDescriptor(deviceRpcResponseProtoSchema), + protoTransportPayloadConfiguration.getRpcRequestDynamicMessageBuilder(deviceRpcRequestProtoSchema) + ); } } else { throw new AdaptorException("Invalid CoapDeviceTypeConfiguration type: " + coapDeviceTypeConfiguration.getClass().getSimpleName() + "!"); @@ -567,11 +579,15 @@ public class CoapTransportResource extends AbstractCoapTransportResource { private boolean jsonPayload; private Descriptors.Descriptor telemetryMsgDescriptor; private Descriptors.Descriptor attributesMsgDescriptor; + private Descriptors.Descriptor rpcResponseMsgDescriptor; + private DynamicMessage.Builder rpcRequestDynamicMessageBuilder; - public TransportConfigurationContainer(boolean jsonPayload, Descriptors.Descriptor telemetryMsgDescriptor, Descriptors.Descriptor attributesMsgDescriptor) { + public TransportConfigurationContainer(boolean jsonPayload, Descriptors.Descriptor telemetryMsgDescriptor, Descriptors.Descriptor attributesMsgDescriptor, Descriptors.Descriptor rpcResponseMsgDescriptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) { this.jsonPayload = jsonPayload; this.telemetryMsgDescriptor = telemetryMsgDescriptor; this.attributesMsgDescriptor = attributesMsgDescriptor; + this.rpcResponseMsgDescriptor = rpcResponseMsgDescriptor; + this.rpcRequestDynamicMessageBuilder = rpcRequestDynamicMessageBuilder; } public TransportConfigurationContainer(boolean jsonPayload) { diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java index 1d4703ead4..b30c86307b 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java @@ -16,6 +16,7 @@ package org.thingsboard.server.transport.coap.adaptors; import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; import org.eclipse.californium.core.coap.Request; import org.eclipse.californium.core.coap.Response; import org.thingsboard.server.common.transport.adaptor.AdaptorException; @@ -32,7 +33,7 @@ public interface CoapTransportAdaptor { TransportProtos.GetAttributeRequestMsg convertToGetAttributes(UUID sessionId, Request inbound) throws AdaptorException; - TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound) throws AdaptorException; + TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound, Descriptors.Descriptor rpcResponseMsgDescriptor) throws AdaptorException; TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(UUID sessionId, Request inbound) throws AdaptorException; @@ -42,7 +43,7 @@ public interface CoapTransportAdaptor { Response convertToPublish(boolean isConfirmable, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException; - Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException; + Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException; Response convertToPublish(TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException; diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java index e382b2a3ee..85d35301c3 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java @@ -20,6 +20,7 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import com.google.gson.JsonSyntaxException; import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; import lombok.extern.slf4j.Slf4j; import org.eclipse.californium.core.coap.CoAP; import org.eclipse.californium.core.coap.Request; @@ -64,7 +65,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { } @Override - public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound) throws AdaptorException { + public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound, Descriptors.Descriptor rpcResponseMsgDescriptor) throws AdaptorException { Optional requestId = CoapTransportResource.getRequestId(inbound); String payload = validatePayload(sessionId, inbound, false); JsonObject response = new JsonParser().parse(payload).getAsJsonObject(); @@ -95,7 +96,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg msg) throws AdaptorException { + public Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg msg, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { return getObserveNotification(isConfirmable, JsonConverter.toJson(msg, true)); } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java index d4c53fdc9a..120e49f697 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.transport.coap.adaptors; +import com.google.gson.JsonElement; import com.google.gson.JsonParser; import com.google.protobuf.Descriptors; import com.google.protobuf.DynamicMessage; @@ -63,16 +64,16 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor { } @Override - public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound) throws AdaptorException { + public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound, Descriptors.Descriptor rpcResponseMsgDescriptor) throws AdaptorException { Optional requestId = CoapTransportResource.getRequestId(inbound); if (requestId.isEmpty()) { throw new AdaptorException("Request id is missing!"); } else { try { - String payload = TransportProtos.ToDeviceRpcResponseMsg.parseFrom(inbound.getPayload()).getPayload(); - return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId.get()) - .setPayload(payload).build(); - } catch (InvalidProtocolBufferException e) { + JsonElement response = new JsonParser().parse(dynamicMsgToJson(inbound.getPayload(), rpcResponseMsgDescriptor)); + return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId.orElseThrow(() -> new AdaptorException("Request id is missing!"))) + .setPayload(response.toString()).build(); + } catch (Exception e) { throw new AdaptorException(e); } } @@ -112,8 +113,8 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor { } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg msg) throws AdaptorException { - return getObserveNotification(isConfirmable, msg.toByteArray()); + public Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { + return getObserveNotification(isConfirmable, ProtoConverter.convertToRpcRequest(rpcRequest, rpcRequestDynamicMessageBuilder)); } @Override diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java index c948e755c2..30178bd44f 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.transport.mqtt.adaptors; +import com.google.gson.JsonElement; import com.google.gson.JsonParser; import com.google.protobuf.Descriptors; import com.google.protobuf.DynamicMessage; @@ -64,9 +65,9 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor { public TransportProtos.PostAttributeMsg convertToPostAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException { DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx; byte[] bytes = toBytes(inbound.payload()); - Descriptors.Descriptor attributesDynamicMessage = getDescriptor(deviceSessionCtx.getAttributesDynamicMessageDescriptor()); + Descriptors.Descriptor attributesDynamicMessageDescriptor = getDescriptor(deviceSessionCtx.getAttributesDynamicMessageDescriptor()); try { - return JsonConverter.convertToAttributesProto(new JsonParser().parse(dynamicMsgToJson(bytes, attributesDynamicMessage))); + return JsonConverter.convertToAttributesProto(new JsonParser().parse(dynamicMsgToJson(bytes, attributesDynamicMessageDescriptor))); } catch (Exception e) { throw new AdaptorException(e); } @@ -86,8 +87,8 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor { public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException { byte[] bytes = toBytes(inbound.payload()); String topicName = inbound.variableHeader().topicName(); - int requestId = getRequestId(topicName, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX); try { + int requestId = getRequestId(topicName, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX); return ProtoConverter.convertToGetAttributeRequestMessage(bytes, requestId); } catch (InvalidProtocolBufferException e) { log.warn("Failed to decode get attributes request", e); @@ -97,10 +98,15 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor { @Override public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException { + DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx; + String topicName = mqttMsg.variableHeader().topicName(); byte[] bytes = toBytes(mqttMsg.payload()); + Descriptors.Descriptor rpcResponseDynamicMessageDescriptor = getDescriptor(deviceSessionCtx.getRpcResponseDynamicMessageDescriptor()); try { - return TransportProtos.ToDeviceRpcResponseMsg.parseFrom(bytes); - } catch (RuntimeException | InvalidProtocolBufferException e) { + int requestId = getRequestId(topicName, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC); + JsonElement response = new JsonParser().parse(dynamicMsgToJson(bytes, rpcResponseDynamicMessageDescriptor)); + return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId).setPayload(response.toString()).build(); + } catch (Exception e) { log.warn("Failed to decode Rpc response", e); throw new AdaptorException(e); } @@ -145,8 +151,10 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor { @Override - public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) { - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), ProtoConverter.convertToRpcRequest(rpcRequest))); + public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException { + DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx; + DynamicMessage.Builder rpcRequestDynamicMessageBuilder = deviceSessionCtx.getRpcRequestDynamicMessageBuilder(); + return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), ProtoConverter.convertToRpcRequest(rpcRequest, rpcRequestDynamicMessageBuilder))); } @Override diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java index efc5c057ed..3804d96cf6 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java @@ -16,6 +16,7 @@ package org.thingsboard.server.transport.mqtt.session; import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; import io.netty.channel.ChannelHandlerContext; import lombok.Getter; import lombok.Setter; @@ -60,6 +61,8 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext { private volatile TransportPayloadType payloadType = TransportPayloadType.JSON; private volatile Descriptors.Descriptor attributesDynamicMessageDescriptor; private volatile Descriptors.Descriptor telemetryDynamicMessageDescriptor; + private volatile Descriptors.Descriptor rpcResponseDynamicMessageDescriptor; + private volatile DynamicMessage.Builder rpcRequestDynamicMessageBuilder; @Getter @Setter @@ -102,6 +105,14 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext { return attributesDynamicMessageDescriptor; } + public Descriptors.Descriptor getRpcResponseDynamicMessageDescriptor() { + return rpcResponseDynamicMessageDescriptor; + } + + public DynamicMessage.Builder getRpcRequestDynamicMessageBuilder() { + return rpcRequestDynamicMessageBuilder; + } + @Override public void setDeviceProfile(DeviceProfile deviceProfile) { super.setDeviceProfile(deviceProfile); @@ -136,5 +147,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext { ProtoTransportPayloadConfiguration protoTransportPayloadConfig = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration; telemetryDynamicMessageDescriptor = protoTransportPayloadConfig.getTelemetryDynamicMessageDescriptor(protoTransportPayloadConfig.getDeviceTelemetryProtoSchema()); attributesDynamicMessageDescriptor = protoTransportPayloadConfig.getAttributesDynamicMessageDescriptor(protoTransportPayloadConfig.getDeviceAttributesProtoSchema()); + rpcResponseDynamicMessageDescriptor = protoTransportPayloadConfig.getRpcResponseDynamicMessageDescriptor(protoTransportPayloadConfig.getDeviceRpcResponseProtoSchema()); + rpcRequestDynamicMessageBuilder = protoTransportPayloadConfig.getRpcRequestDynamicMessageBuilder(protoTransportPayloadConfig.getDeviceRpcRequestProtoSchema()); } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/AdaptorException.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/AdaptorException.java index 42c99bc428..d0d471a784 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/AdaptorException.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/AdaptorException.java @@ -31,4 +31,8 @@ public class AdaptorException extends Exception { super(cause); } + public AdaptorException(String message, Exception cause) { + super(message, cause); + } + } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/ProtoConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/ProtoConverter.java index ed440ebcfe..620728cda4 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/ProtoConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/ProtoConverter.java @@ -15,10 +15,13 @@ */ package org.thingsboard.server.common.transport.adaptor; +import com.google.gson.Gson; import com.google.gson.JsonElement; +import com.google.gson.JsonObject; import com.google.gson.JsonParser; -import com.google.gson.JsonPrimitive; +import com.google.protobuf.DynamicMessage; import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.util.JsonFormat; import lombok.extern.slf4j.Slf4j; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -34,6 +37,7 @@ import java.util.List; @Slf4j public class ProtoConverter { + public static final Gson GSON = new Gson(); public static final JsonParser JSON_PARSER = new JsonParser(); public static TransportProtos.PostTelemetryMsg convertToTelemetryProto(byte[] payload) throws InvalidProtocolBufferException, IllegalArgumentException { @@ -170,26 +174,20 @@ public class ProtoConverter { return kvList; } - public static byte[] convertToRpcRequest(TransportProtos.ToDeviceRpcRequestMsg toDeviceRpcRequestMsg) { - TransportProtos.ToDeviceRpcRequestMsg.Builder toDeviceRpcRequestMsgBuilder = toDeviceRpcRequestMsg.newBuilderForType(); - toDeviceRpcRequestMsgBuilder.mergeFrom(toDeviceRpcRequestMsg); - toDeviceRpcRequestMsgBuilder.setParams(parseParams(toDeviceRpcRequestMsg)); - TransportProtos.ToDeviceRpcRequestMsg result = toDeviceRpcRequestMsgBuilder.build(); - return result.toByteArray(); - } - - private static String parseParams(TransportProtos.ToDeviceRpcRequestMsg toDeviceRpcRequestMsg) { + public static byte[] convertToRpcRequest(TransportProtos.ToDeviceRpcRequestMsg toDeviceRpcRequestMsg, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { + rpcRequestDynamicMessageBuilder.clear(); + JsonObject rpcRequestJson = new JsonObject(); + rpcRequestJson.addProperty("method", toDeviceRpcRequestMsg.getMethodName()); + rpcRequestJson.addProperty("requestId", toDeviceRpcRequestMsg.getRequestId()); String params = toDeviceRpcRequestMsg.getParams(); - JsonElement jsonElementParams = JSON_PARSER.parse(params); - if (!jsonElementParams.isJsonPrimitive()) { - return params; - } else { - JsonPrimitive primitiveParams = jsonElementParams.getAsJsonPrimitive(); - if (jsonElementParams.getAsJsonPrimitive().isString()) { - return primitiveParams.getAsString(); - } else { - return params; - } + try { + JsonElement paramsElement = JSON_PARSER.parse(params); + rpcRequestJson.add("params", paramsElement); + JsonFormat.parser().ignoringUnknownFields().merge(GSON.toJson(rpcRequestJson), rpcRequestDynamicMessageBuilder); + DynamicMessage dynamicRpcRequest = rpcRequestDynamicMessageBuilder.build(); + return dynamicRpcRequest.toByteArray(); + } catch (Exception e) { + throw new AdaptorException("Failed to convert ToDeviceRpcRequestMsg to Dynamic Rpc request message due to: ", e); } } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java index bff523d956..48879d2f15 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.dao.device; +import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; import com.squareup.wire.Syntax; import com.squareup.wire.schema.Field; import com.squareup.wire.schema.Location; @@ -87,6 +89,8 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D private static final Location LOCATION = new Location("", "", -1, -1); private static final String ATTRIBUTES_PROTO_SCHEMA = "attributes proto schema"; private static final String TELEMETRY_PROTO_SCHEMA = "telemetry proto schema"; + private static final String RPC_REQUEST_PROTO_SCHEMA = "rpc request proto schema"; + private static final String RPC_RESPONSE_PROTO_SCHEMA = "rpc response proto schema"; private static String invalidSchemaProvidedMessage(String schemaName) { return "[Transport Configuration] invalid " + schemaName + " provided!"; @@ -357,9 +361,10 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D if (transportConfiguration instanceof MqttDeviceProfileTransportConfiguration) { MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration; if (mqttTransportConfiguration.getTransportPayloadTypeConfiguration() instanceof ProtoTransportPayloadConfiguration) { - ProtoTransportPayloadConfiguration protoTransportPayloadTypeConfiguration = + ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) mqttTransportConfiguration.getTransportPayloadTypeConfiguration(); - validateProtoSchemas(protoTransportPayloadTypeConfiguration); + validateProtoSchemas(protoTransportPayloadConfiguration); + validateRpcRequestDynamicMessageFields(protoTransportPayloadConfiguration); } } else if (transportConfiguration instanceof CoapDeviceProfileTransportConfiguration) { CoapDeviceProfileTransportConfiguration coapDeviceProfileTransportConfiguration = (CoapDeviceProfileTransportConfiguration) transportConfiguration; @@ -370,6 +375,7 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D if (transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration) { ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration; validateProtoSchemas(protoTransportPayloadConfiguration); + validateRpcRequestDynamicMessageFields(protoTransportPayloadConfiguration); } } } @@ -417,6 +423,8 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D try { validateTransportProtoSchema(protoTransportPayloadTypeConfiguration.getDeviceAttributesProtoSchema(), ATTRIBUTES_PROTO_SCHEMA); validateTransportProtoSchema(protoTransportPayloadTypeConfiguration.getDeviceTelemetryProtoSchema(), TELEMETRY_PROTO_SCHEMA); + validateTransportProtoSchema(protoTransportPayloadTypeConfiguration.getDeviceRpcRequestProtoSchema(), RPC_REQUEST_PROTO_SCHEMA); + validateTransportProtoSchema(protoTransportPayloadTypeConfiguration.getDeviceRpcResponseProtoSchema(), RPC_RESPONSE_PROTO_SCHEMA); } catch (Exception exception) { throw new DataValidationException(exception.getMessage()); } @@ -539,6 +547,48 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D }; + private void validateRpcRequestDynamicMessageFields(ProtoTransportPayloadConfiguration protoTransportPayloadTypeConfiguration) { + DynamicMessage.Builder rpcRequestDynamicMessageBuilder = protoTransportPayloadTypeConfiguration.getRpcRequestDynamicMessageBuilder(protoTransportPayloadTypeConfiguration.getDeviceRpcRequestProtoSchema()); + Descriptors.Descriptor rpcRequestDynamicMessageDescriptor = rpcRequestDynamicMessageBuilder.getDescriptorForType(); + if (rpcRequestDynamicMessageDescriptor == null) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Failed to get rpcRequestDynamicMessageDescriptor!"); + } else { + if (CollectionUtils.isEmpty(rpcRequestDynamicMessageDescriptor.getFields()) || rpcRequestDynamicMessageDescriptor.getFields().size() != 3) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " " + rpcRequestDynamicMessageDescriptor.getName() + " message should always contains 3 fields: method, requestId and params!"); + } + Descriptors.FieldDescriptor methodFieldDescriptor = rpcRequestDynamicMessageDescriptor.findFieldByName("method"); + if (methodFieldDescriptor == null) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Failed to get field descriptor for field: method!"); + } else { + if (!Descriptors.FieldDescriptor.Type.STRING.equals(methodFieldDescriptor.getType())) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Field 'method' has invalid data type. Only string type is supported!"); + } + if (methodFieldDescriptor.isRepeated()) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Field 'method' has invalid label!"); + } + } + Descriptors.FieldDescriptor requestIdFieldDescriptor = rpcRequestDynamicMessageDescriptor.findFieldByName("requestId"); + if (requestIdFieldDescriptor == null) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Failed to get field descriptor for field: requestId!"); + } else { + if (!Descriptors.FieldDescriptor.Type.INT32.equals(requestIdFieldDescriptor.getType())) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Field 'requestId' has invalid data type. Only int32 type is supported!"); + } + if (requestIdFieldDescriptor.isRepeated()) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Field 'requestId' has invalid label!"); + } + } + Descriptors.FieldDescriptor paramsFieldDescriptor = rpcRequestDynamicMessageDescriptor.findFieldByName("params"); + if (paramsFieldDescriptor == null) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + " Failed to get field descriptor for field: params!"); + } else { + if (paramsFieldDescriptor.isRepeated()) { + throw new DataValidationException(invalidSchemaProvidedMessage(RPC_REQUEST_PROTO_SCHEMA) + "Field 'params' has invalid label!"); + } + } + } + } + private PaginatedRemover tenantDeviceProfilesRemover = new PaginatedRemover() { diff --git a/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.html b/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.html index e04ecfb258..320373e52c 100644 --- a/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.html +++ b/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.html @@ -61,6 +61,21 @@ {{ 'device-profile.attributes-proto-schema-required' | translate}} + + device-profile.rpc-request-proto-schema + + + {{ 'device-profile.rpc-request-proto-schema-required' | translate}} + + device-profile.rpc-request-proto-schema-hint + + + device-profile.rpc-response-proto-schema + + + {{ 'device-profile.rpc-response-proto-schema-required' | translate}} + + diff --git a/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.ts b/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.ts index 4a686d39bd..dc1da3d973 100644 --- a/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.ts +++ b/ui-ngx/src/app/modules/home/components/profile/device/coap-device-profile-transport-configuration.component.ts @@ -24,6 +24,8 @@ import { coapDeviceTypeTranslationMap, CoapTransportDeviceType, defaultAttributesSchema, + defaultRpcRequestSchema, + defaultRpcResponseSchema, defaultTelemetrySchema, DeviceProfileTransportConfiguration, DeviceTransportType, @@ -59,7 +61,9 @@ export class CoapDeviceProfileTransportConfigurationComponent implements Control private transportPayloadTypeConfiguration = this.fb.group({ transportPayloadType: [TransportPayloadType.JSON, Validators.required], deviceTelemetryProtoSchema: [defaultTelemetrySchema, Validators.required], - deviceAttributesProtoSchema: [defaultAttributesSchema, Validators.required] + deviceAttributesProtoSchema: [defaultAttributesSchema, Validators.required], + deviceRpcRequestProtoSchema: [defaultRpcRequestSchema, Validators.required], + deviceRpcResponseProtoSchema: [defaultRpcResponseSchema, Validators.required] }); get required(): boolean { diff --git a/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html b/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html index 670e6fbb72..55f0869dbc 100644 --- a/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html +++ b/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html @@ -88,6 +88,21 @@ {{ 'device-profile.attributes-proto-schema-required' | translate}} + + device-profile.rpc-request-proto-schema + + + {{ 'device-profile.rpc-request-proto-schema-required' | translate}} + + device-profile.rpc-request-proto-schema-hint + + + device-profile.rpc-response-proto-schema + + + {{ 'device-profile.rpc-response-proto-schema-required' | translate}} + + diff --git a/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts b/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts index a61ad841aa..381e2f9e8d 100644 --- a/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts +++ b/ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts @@ -29,6 +29,8 @@ import { AppState } from '@app/core/core.state'; import { coerceBooleanProperty } from '@angular/cdk/coercion'; import { defaultAttributesSchema, + defaultRpcRequestSchema, + defaultRpcResponseSchema, defaultTelemetrySchema, DeviceProfileTransportConfiguration, DeviceTransportType, @@ -90,7 +92,9 @@ export class MqttDeviceProfileTransportConfigurationComponent implements Control transportPayloadTypeConfiguration: this.fb.group({ transportPayloadType: [TransportPayloadType.JSON, Validators.required], deviceTelemetryProtoSchema: [defaultTelemetrySchema, Validators.required], - deviceAttributesProtoSchema: [defaultAttributesSchema, Validators.required] + deviceAttributesProtoSchema: [defaultAttributesSchema, Validators.required], + deviceRpcRequestProtoSchema: [defaultRpcRequestSchema, Validators.required], + deviceRpcResponseProtoSchema: [defaultRpcResponseSchema, Validators.required] }) }, {validator: this.uniqueDeviceTopicValidator} ); @@ -139,15 +143,21 @@ export class MqttDeviceProfileTransportConfigurationComponent implements Control if (forceUpdated) { transportPayloadTypeForm.patchValue({ deviceTelemetryProtoSchema: defaultTelemetrySchema, - deviceAttributesProtoSchema: defaultAttributesSchema + deviceAttributesProtoSchema: defaultAttributesSchema, + deviceRpcRequestProtoSchema: defaultRpcRequestSchema, + deviceRpcResponseProtoSchema: defaultRpcResponseSchema }, {emitEvent: false}); } if (type === TransportPayloadType.PROTOBUF && !this.disabled) { transportPayloadTypeForm.get('deviceTelemetryProtoSchema').enable({emitEvent: false}); transportPayloadTypeForm.get('deviceAttributesProtoSchema').enable({emitEvent: false}); + transportPayloadTypeForm.get('deviceRpcRequestProtoSchema').enable({emitEvent: false}); + transportPayloadTypeForm.get('deviceRpcResponseProtoSchema').enable({emitEvent: false}); } else { transportPayloadTypeForm.get('deviceTelemetryProtoSchema').disable({emitEvent: false}); transportPayloadTypeForm.get('deviceAttributesProtoSchema').disable({emitEvent: false}); + transportPayloadTypeForm.get('deviceRpcRequestProtoSchema').enable({emitEvent: false}); + transportPayloadTypeForm.get('deviceRpcResponseProtoSchema').disable({emitEvent: false}); } } diff --git a/ui-ngx/src/app/shared/models/device.models.ts b/ui-ngx/src/app/shared/models/device.models.ts index 52990127bf..46b457a2ee 100644 --- a/ui-ngx/src/app/shared/models/device.models.ts +++ b/ui-ngx/src/app/shared/models/device.models.ts @@ -140,6 +140,24 @@ export const defaultAttributesSchema = ' string serialNumber = 2;\n' + '}'; +export const defaultRpcRequestSchema = + 'syntax ="proto3";\n' + + 'package rpc;\n' + + '\n' + + 'message RpcRequestMsg {\n' + + ' string method = 1;\n' + + ' int32 requestId = 2;\n' + + ' string params = 3;\n' + + '}'; + +export const defaultRpcResponseSchema = + 'syntax ="proto3";\n' + + 'package rpc;\n' + + '\n' + + 'message RpcResponseMsg {\n' + + ' string payload = 1;\n' + + '}'; + export const coapDeviceTypeTranslationMap = new Map( [ [CoapTransportDeviceType.DEFAULT, 'device-profile.coap-device-type-default'], @@ -170,6 +188,13 @@ export const deviceTransportTypeConfigurationInfoMap = new Map