tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #6240 from ShvaykaD/feature/6233 

[3.4] MQTT added ability to send PUBACK instead of close session on publish request data validation exception
pull/6485/head
Andrew Shvayka 4 years ago
committed by GitHub
parent
00de756e03 d4e6c013d2
commit
84dc8881ba
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 302 additions and 71 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 10
      application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
  2. 57
      application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java
  3. 2
      application/src/test/java/org/thingsboard/server/controller/BaseOtaPackageControllerTest.java
  4. 2
      application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java
  5. 24
      application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java
  6. 8
      application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestBackwardCompatibilityIntegrationTest.java
  7. 6
      application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java
  8. 14
      application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java
  9. 14
      application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java
  10. 20
      application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcBackwardCompatibilityIntegrationTest.java
  11. 2
      application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java
  12. 33
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java
  13. 66
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java
  14. 85
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java
  15. 1
      common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java
  16. 11
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
  17. 7
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java
  18. 6
      ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html
  19. 1
      ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts
  20. 2
      ui-ngx/src/app/shared/models/device.models.ts
  21. 2
      ui-ngx/src/assets/locale/locale.constant-en_US.json

10
application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
View File

@ -377,18 +377,23 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
}
}
protected DeviceProfile createDeviceProfile(String name) {
return createDeviceProfile(name, null);
}
protected DeviceProfile createDeviceProfile(String name, DeviceProfileTransportConfiguration deviceProfileTransportConfiguration) {
DeviceProfile deviceProfile = new DeviceProfile();
deviceProfile.setName(name);
deviceProfile.setType(DeviceProfileType.DEFAULT);
deviceProfile.setTransportType(DeviceTransportType.DEFAULT);
deviceProfile.setDescription(name + " Test");
DeviceProfileData deviceProfileData = new DeviceProfileData();
DefaultDeviceProfileConfiguration configuration = new DefaultDeviceProfileConfiguration();
deviceProfileData.setConfiguration(configuration);
if (deviceProfileTransportConfiguration != null) {
deviceProfile.setTransportType(deviceProfileTransportConfiguration.getType());
deviceProfileData.setTransportConfiguration(deviceProfileTransportConfiguration);
} else {
deviceProfile.setTransportType(DeviceTransportType.DEFAULT);
deviceProfileData.setTransportConfiguration(new DefaultDeviceProfileTransportConfiguration());
}
deviceProfile.setProfileData(deviceProfileData);
@ -397,10 +402,11 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
return deviceProfile;
}
protected MqttDeviceProfileTransportConfiguration createMqttDeviceProfileTransportConfiguration(TransportPayloadTypeConfiguration transportPayloadTypeConfiguration) {
protected MqttDeviceProfileTransportConfiguration createMqttDeviceProfileTransportConfiguration(TransportPayloadTypeConfiguration transportPayloadTypeConfiguration, boolean sendAckOnValidationException) {
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = new MqttDeviceProfileTransportConfiguration();
mqttDeviceProfileTransportConfiguration.setDeviceTelemetryTopic(MqttTopics.DEVICE_TELEMETRY_TOPIC);
mqttDeviceProfileTransportConfiguration.setDeviceTelemetryTopic(MqttTopics.DEVICE_ATTRIBUTES_TOPIC);
mqttDeviceProfileTransportConfiguration.setSendAckOnValidationException(sendAckOnValidationException);
mqttDeviceProfileTransportConfiguration.setTransportPayloadTypeConfiguration(transportPayloadTypeConfiguration);
return mqttDeviceProfileTransportConfiguration;
}

57
application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java
View File

@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.DeviceTransportType;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.device.profile.DeviceProfileTransportConfiguration;
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;
@ -93,7 +94,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testSaveDeviceProfile() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
Assert.assertNotNull(savedDeviceProfile);
Assert.assertNotNull(savedDeviceProfile.getId());
@ -112,13 +113,13 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void saveDeviceProfileWithViolationOfValidation() throws Exception {
doPost("/api/deviceProfile", this.createDeviceProfile(RandomStringUtils.randomAlphabetic(300), null))
doPost("/api/deviceProfile", this.createDeviceProfile(RandomStringUtils.randomAlphabetic(300)))
.andExpect(statusReason(containsString("length of name must be equal or less than 255")));
}
@Test
public void testFindDeviceProfileById() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
DeviceProfile foundDeviceProfile = doGet("/api/deviceProfile/"+savedDeviceProfile.getId().getId().toString(), DeviceProfile.class);
Assert.assertNotNull(foundDeviceProfile);
@ -127,7 +128,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testFindDeviceProfileInfoById() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
DeviceProfileInfo foundDeviceProfileInfo = doGet("/api/deviceProfileInfo/"+savedDeviceProfile.getId().getId().toString(), DeviceProfileInfo.class);
Assert.assertNotNull(foundDeviceProfileInfo);
@ -149,7 +150,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testSetDefaultDeviceProfile() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile 1", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile 1");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
DeviceProfile defaultDeviceProfile = doPost("/api/deviceProfile/"+savedDeviceProfile.getId().getId().toString()+"/default", DeviceProfile.class);
Assert.assertNotNull(defaultDeviceProfile);
@ -169,19 +170,19 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testSaveDeviceProfileWithSameName() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
doPost("/api/deviceProfile", deviceProfile).andExpect(status().isOk());
DeviceProfile deviceProfile2 = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile2 = this.createDeviceProfile("Device Profile");
doPost("/api/deviceProfile", deviceProfile2).andExpect(status().isBadRequest())
.andExpect(statusReason(containsString("Device profile with such name already exists")));
}
@Test
public void testSaveDeviceProfileWithSameProvisionDeviceKey() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
deviceProfile.setProvisionDeviceKey("testProvisionDeviceKey");
doPost("/api/deviceProfile", deviceProfile).andExpect(status().isOk());
DeviceProfile deviceProfile2 = this.createDeviceProfile("Device Profile 2", null);
DeviceProfile deviceProfile2 = this.createDeviceProfile("Device Profile 2");
deviceProfile2.setProvisionDeviceKey("testProvisionDeviceKey");
doPost("/api/deviceProfile", deviceProfile2).andExpect(status().isBadRequest())
.andExpect(statusReason(containsString("Device profile with such provision device key already exists")));
@ -190,7 +191,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Ignore
@Test
public void testChangeDeviceProfileTypeWithExistingDevices() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
Device device = new Device();
device.setName("Test device");
@ -205,7 +206,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testChangeDeviceProfileTransportTypeWithExistingDevices() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
Device device = new Device();
device.setName("Test device");
@ -219,7 +220,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testDeleteDeviceProfileWithExistingDevice() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
Device device = new Device();
@ -236,7 +237,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
@Test
public void testDeleteDeviceProfile() throws Exception {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
doDelete("/api/deviceProfile/" + savedDeviceProfile.getId().getId().toString())
@ -257,7 +258,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
deviceProfiles.addAll(pageData.getData());
for (int i=0;i<28;i++) {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile"+i, null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile"+i);
deviceProfiles.add(doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class));
}
@ -302,7 +303,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
deviceProfiles.addAll(deviceProfilePageData.getData());
for (int i=0;i<28;i++) {
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile"+i, null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile"+i);
deviceProfiles.add(doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class));
}
@ -835,19 +836,35 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
"}", "[Transport Configuration] invalid rpc request proto schema provided! Failed to get field descriptor for field: params!");
}
@Test
public void testSaveDeviceProfileWithSendAckOnValidationException() throws Exception {
JsonTransportPayloadConfiguration jsonTransportPayloadConfiguration = new JsonTransportPayloadConfiguration();
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(jsonTransportPayloadConfiguration, true);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", mqttDeviceProfileTransportConfiguration);
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
Assert.assertNotNull(savedDeviceProfile);
Assert.assertEquals(savedDeviceProfile.getTransportType(), DeviceTransportType.MQTT);
Assert.assertTrue(savedDeviceProfile.getProfileData().getTransportConfiguration() instanceof MqttDeviceProfileTransportConfiguration);
MqttDeviceProfileTransportConfiguration transportConfiguration = (MqttDeviceProfileTransportConfiguration) savedDeviceProfile.getProfileData().getTransportConfiguration();
Assert.assertTrue(transportConfiguration.isSendAckOnValidationException());
DeviceProfile foundDeviceProfile = doGet("/api/deviceProfile/"+ savedDeviceProfile.getId().getId().toString(), DeviceProfile.class);
Assert.assertEquals(savedDeviceProfile, foundDeviceProfile);
}
private DeviceProfile testSaveDeviceProfileWithProtoPayloadType(String schema) throws Exception {
ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = this.createProtoTransportPayloadConfiguration(schema, schema, null, null);
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(protoTransportPayloadConfiguration);
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(protoTransportPayloadConfiguration, false);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", mqttDeviceProfileTransportConfiguration);
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
DeviceProfile foundDeviceProfile = doGet("/api/deviceProfile/"+savedDeviceProfile.getId().getId().toString(), DeviceProfile.class);
Assert.assertEquals(savedDeviceProfile.getName(), foundDeviceProfile.getName());
Assert.assertNotNull(savedDeviceProfile);
DeviceProfile foundDeviceProfile = doGet("/api/deviceProfile/"+ savedDeviceProfile.getId().getId().toString(), DeviceProfile.class);
Assert.assertEquals(savedDeviceProfile, foundDeviceProfile);
return savedDeviceProfile;
}
private void testSaveDeviceProfileWithInvalidProtoSchema(String schema, String errorMsg) throws Exception {
ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = this.createProtoTransportPayloadConfiguration(schema, schema, null, null);
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(protoTransportPayloadConfiguration);
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(protoTransportPayloadConfiguration, false);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", mqttDeviceProfileTransportConfiguration);
doPost("/api/deviceProfile", deviceProfile).andExpect(status().isBadRequest())
.andExpect(statusReason(containsString(errorMsg)));
@ -855,7 +872,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
private void testSaveDeviceProfileWithInvalidRpcRequestProtoSchema(String schema, String errorMsg) throws Exception {
ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = this.createProtoTransportPayloadConfiguration(schema, schema, schema, null);
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(protoTransportPayloadConfiguration);
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(protoTransportPayloadConfiguration, false);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", mqttDeviceProfileTransportConfiguration);
doPost("/api/deviceProfile", deviceProfile).andExpect(status().isBadRequest())
.andExpect(statusReason(containsString(errorMsg)));

2
application/src/test/java/org/thingsboard/server/controller/BaseOtaPackageControllerTest.java
View File

@ -79,7 +79,7 @@ public abstract class BaseOtaPackageControllerTest extends AbstractControllerTes
tenantAdmin = createUserAndLogin(tenantAdmin, "testPassword1");
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", null);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile");
DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
Assert.assertNotNull(savedDeviceProfile);
deviceProfileId = savedDeviceProfile.getId();

2
application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java
View File

@ -199,7 +199,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
private void installation() throws Exception {
edge = doPost("/api/edge", constructEdge("Test Edge", "test"), Edge.class);
DeviceProfile deviceProfile = this.createDeviceProfile(CUSTOM_DEVICE_PROFILE_NAME, null);
DeviceProfile deviceProfile = this.createDeviceProfile(CUSTOM_DEVICE_PROFILE_NAME);
extendDeviceProfileData(deviceProfile);
doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);

24
application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java
View File

@ -66,11 +66,19 @@ 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false);
this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
}
protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic, boolean enableCompatibilityWithJsonPayloadFormat, boolean useJsonPayloadFormatForDefaultDownlinkTopics) throws Exception {
this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics);
this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics, false);
}
protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic, boolean enableCompatibilityWithJsonPayloadFormat, boolean useJsonPayloadFormatForDefaultDownlinkTopics, boolean sendAckOnValidationException) throws Exception {
this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics, sendAckOnValidationException);
}
protected void processBeforeTest(String deviceName, String gatewayName, TransportPayloadType payloadType, String telemetryTopic, String attributesTopic, boolean sendAckOnValidationException) throws Exception {
this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, sendAckOnValidationException);
}
protected void processBeforeTest(String deviceName,
@ -86,7 +94,8 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
String provisionSecret,
DeviceProfileProvisionType provisionType,
boolean enableCompatibilityWithJsonPayloadFormat,
boolean useJsonPayloadFormatForDefaultDownlinkTopics) throws Exception {
boolean useJsonPayloadFormatForDefaultDownlinkTopics,
boolean sendAckOnValidationException) throws Exception {
loginSysAdmin();
Tenant tenant = new Tenant();
@ -115,7 +124,10 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
gateway.setAdditionalInfo(additionalInfo);
if (payloadType != null) {
DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic, telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema, provisionKey, provisionSecret, provisionType, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics);
DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic,
telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema,
provisionKey, provisionSecret, provisionType, enableCompatibilityWithJsonPayloadFormat,
useJsonPayloadFormatForDefaultDownlinkTopics, sendAckOnValidationException);
deviceProfile = doPost("/api/deviceProfile", mqttDeviceProfile, DeviceProfile.class);
device.setType(deviceProfile.getName());
device.setDeviceProfileId(deviceProfile.getId());
@ -173,7 +185,8 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
String provisionKey, String provisionSecret,
DeviceProfileProvisionType provisionType,
boolean enableCompatibilityWithJsonPayloadFormat,
boolean useJsonPayloadFormatForDefaultDownlinkTopics) {
boolean useJsonPayloadFormatForDefaultDownlinkTopics,
boolean sendAckOnValidationException) {
DeviceProfile deviceProfile = new DeviceProfile();
deviceProfile.setName(transportPayloadType.name());
deviceProfile.setType(DeviceProfileType.DEFAULT);
@ -190,6 +203,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
if (!StringUtils.isEmpty(attributesTopic)) {
mqttDeviceProfileTransportConfiguration.setDeviceAttributesTopic(attributesTopic);
}
mqttDeviceProfileTransportConfiguration.setSendAckOnValidationException(sendAckOnValidationException);
TransportPayloadTypeConfiguration transportPayloadTypeConfiguration;
if (TransportPayloadType.JSON.equals(transportPayloadType)) {
transportPayloadTypeConfiguration = new JsonTransportPayloadConfiguration();

8
application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestBackwardCompatibilityIntegrationTest.java
View File

@ -38,28 +38,28 @@ public abstract class AbstractMqttAttributesRequestBackwardCompatibilityIntegrat
@Test
public void testRequestAttributesValuesFromTheServerWithEnabledJsonCompatibility() 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, false);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, false, false);
processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
}
@Test
public void testRequestAttributesValuesFromTheServerWithEnabledJsonCompatibilityAndJsonDownlinks() 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processJsonTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
}
@Test
public void testRequestAttributesValuesFromTheServerOnShortTopicWithEnabledJsonCompatibilityAndJsonDownlinks() 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX);
}
@Test
public void testRequestAttributesValuesFromTheServerOnShortProtoTopicWithEnabledJsonCompatibilityAndJsonDownlinks() 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX);
}

6
application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/AbstractMqttAttributesRequestProtoIntegrationTest.java
View File

@ -38,21 +38,21 @@ 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
}
@Test
public void testRequestAttributesValuesFromTheServerOnShortTopic() 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX);
}
@Test
public void testRequestAttributesValuesFromTheServerOnShortProtoTopic() 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, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false);
TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
processProtoTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX);
}

14
application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java
View File

@ -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, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false);
super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false);
super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false);
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, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false);
super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.JSON, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false);
byte[] result = createMqttClientAndPublish().getPayloadBytes();
JsonObject response = JsonUtils.parse(new String(result)).getAsJsonObject();
Assert.assertEquals("Provision data was not found!", response.get("errorMsg").getAsString());

14
application/src/test/java/org/thingsboard/server/transport/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java
View File

@ -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, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false);
super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null,null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false);
super.processBeforeTest("Test Provision device3", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES, false, false, false);
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, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false);
super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKey", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false);
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, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false);
super.processBeforeTest("Test Provision device", "Test Provision gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, null, "testProvisionKeyOrig", "testProvisionSecret", DeviceProfileProvisionType.CHECK_PRE_PROVISIONED_DEVICES, false, false, false);
ProvisionDeviceResponseMsg response = ProvisionDeviceResponseMsg.parseFrom(createMqttClientAndPublish().getPayloadBytes());
Assert.assertEquals(ProvisionResponseStatus.NOT_FOUND.name(), response.getStatus().toString());
}

20
application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcBackwardCompatibilityIntegrationTest.java
View File

@ -33,61 +33,61 @@ public abstract class AbstractMqttServerSideRpcBackwardCompatibilityIntegrationT
@Test
public void testServerMqttOneWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
}
@Test
public void testServerMqttOneWayRpcOnShortTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
}
@Test
public void testServerMqttOneWayRpcOnShortProtoTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC);
}
@Test
public void testServerMqttTwoWayRpcWithEnabledJsonCompatibility() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, false);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, false, false);
processProtoTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
}
@Test
public void testServerMqttTwoWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processJsonTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
}
@Test
public void testServerMqttTwoWayRpcOnShortTopic() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processProtoTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
}
@Test
public void testServerMqttTwoWayRpcOnShortProtoTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processProtoTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC);
}
@Test
public void testServerMqttTwoWayRpcOnShortJsonTopicWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processJsonTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC);
}
@Test
public void testGatewayServerMqttOneWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processProtoOneWayRpcTestGateway("Gateway Device OneWay RPC Proto");
}
@Test
public void testGatewayServerMqttTwoWayRpcWithEnabledJsonCompatibilityAndJsonDownlinks() throws Exception {
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true);
super.processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, true, true, false);
processProtoTwoWayRpcTestGateway("Gateway Device TwoWay RPC Proto");
}

2
application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java
View File

@ -28,7 +28,7 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst
@Before
public void beforeTest() throws Exception {
processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, false, false);
processBeforeTest("RPC test device", "RPC test gateway", TransportPayloadType.PROTOBUF, null, null, null, null, null, RPC_REQUEST_PROTO_SCHEMA, null, null, DeviceProfileProvisionType.DISABLED, false, false, false);
}
@After

33
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java
View File

@ -23,6 +23,7 @@ import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.After;
import org.junit.Before;
@ -50,6 +51,9 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
protected static final String PAYLOAD_VALUES_STR = "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4," +
" \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}";
protected static final String MALFORMED_JSON_PAYLOAD = "{\"key1\":, \"key2\":true, \"key3\": 3.0, \"key4\": 4," +
" \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}";
@Before
public void beforeTest() throws Exception {
processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", null, null, null);
@ -368,5 +372,34 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
}
}
public static class TestMqttPublishCallback implements MqttCallback {
private final CountDownLatch latch;
private boolean pubAckReceived;
public boolean isPubAckReceived() {
return pubAckReceived;
}
public TestMqttPublishCallback(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void connectionLost(Throwable throwable) {
latch.countDown();
}
@Override
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
pubAckReceived = iMqttDeliveryToken.getResponse().getType() == MqttWireMessage.MESSAGE_TYPE_PUBACK;
latch.countDown();
}
}
}

66
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java
View File

@ -20,14 +20,15 @@ import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
import java.util.Arrays;
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.assertFalse;
import static org.junit.Assert.assertTrue;
@Slf4j
public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends AbstractMqttTimeseriesIntegrationTest {
@ -36,7 +37,7 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
@Before
public void beforeTest() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
//do nothing, processBeforeTest will be invoked in particular test methods with different parameters
}
@After
@ -46,12 +47,14 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
@Test
public void testPushTelemetry() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
processJsonPayloadTelemetryTest(POST_DATA_TELEMETRY_TOPIC, expectedKeys, PAYLOAD_VALUES_STR.getBytes(), false);
}
@Test
public void testPushTelemetryWithTs() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
String payloadStr = "{\"ts\": 10000, \"values\": " + PAYLOAD_VALUES_STR + "}";
List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
processJsonPayloadTelemetryTest(POST_DATA_TELEMETRY_TOPIC, expectedKeys, payloadStr.getBytes(), true);
@ -59,21 +62,74 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
@Test
public void testPushTelemetryOnShortTopic() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
super.testPushTelemetryOnShortTopic();
}
@Test
public void testPushTelemetryWithTsOnShortJsonTopic() throws Exception {
public void testPushTelemetryOnShortJsonTopic() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
super.testPushTelemetryOnShortJsonTopic();
}
@Test
public void testPushTelemetryGateway() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
super.testPushTelemetryGateway();
}
@Test
public void testGatewayConnect() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null);
super.testGatewayConnect();
}
@Test
public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null, true);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertTrue(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorDisabled() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null, false);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertFalse(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendAckOnErrorEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null, true);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertTrue(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendAckOnErrorDisabled() throws Exception {
processBeforeTest("Test Post Telemetry device json payload", "Test Post Telemetry gateway json payload", TransportPayloadType.JSON, POST_DATA_TELEMETRY_TOPIC, null, false);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertFalse(callback.isPubAckReceived());
}
}

85
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java
View File

@ -36,7 +36,10 @@ import org.thingsboard.server.gen.transport.TransportProtos;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@ -44,6 +47,7 @@ import static org.junit.Assert.assertTrue;
public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends AbstractMqttTimeseriesIntegrationTest {
private static final String POST_DATA_TELEMETRY_TOPIC = "proto/telemetry";
private static final String MALFORMED_PROTO_PAYLOAD = "invalid proto payload str";
@Before
@Override
@ -91,7 +95,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
" }\n" +
" }\n" +
"}";
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, false, false);
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, false, false, false);
DynamicSchema telemetrySchema = getDynamicSchema(schemaStr);
DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
@ -193,7 +197,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
" }\n" +
" }\n" +
"}";
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, false, false);
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, false, false, false);
DynamicSchema telemetrySchema = getDynamicSchema(schemaStr);
DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
@ -253,7 +257,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
@Test
public void testPushTelemetryGateway() throws Exception {
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, false, false);
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, false, false, false);
TransportApiProtos.GatewayTelemetryMsg.Builder gatewayTelemetryMsgProtoBuilder = TransportApiProtos.GatewayTelemetryMsg.newBuilder();
List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
String deviceName1 = "Device A";
@ -267,7 +271,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
@Test
public void testGatewayConnect() throws Exception {
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, false, false);
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, false, false, false);
String deviceName = "Device A";
TransportApiProtos.ConnectMsg connectMsgProto = getConnectProto(deviceName);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
@ -280,6 +284,79 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
assertNotNull(device);
}
@Test
public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_PROTO_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertTrue(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorDisabled() throws Exception {
processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, false);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_PROTO_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertFalse(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorEnabledAndBackwardCompatibilityEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, true);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertTrue(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorDisabledAndBackwardCompatibilityEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, false);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_JSON_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertFalse(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendAckOnErrorEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_PROTO_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertTrue(callback.isPubAckReceived());
}
@Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendAckOnErrorDisabled() throws Exception {
processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, false);
CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
TestMqttPublishCallback callback = new TestMqttPublishCallback(latch);
client.setCallback(callback);
publishMqttMsg(client, MALFORMED_PROTO_PAYLOAD.getBytes(), POST_DATA_TELEMETRY_TOPIC);
latch.await(3, TimeUnit.SECONDS);
assertFalse(callback.isPubAckReceived());
}
private DynamicSchema getDynamicSchema(String deviceTelemetryProtoSchema) {
DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);

1
common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java
View File

@ -27,6 +27,7 @@ public class MqttDeviceProfileTransportConfiguration implements DeviceProfileTra
@NoXss
private String deviceAttributesTopic = MqttTopics.DEVICE_ATTRIBUTES_TOPIC;
private TransportPayloadTypeConfiguration transportPayloadTypeConfiguration;
private boolean sendAckOnValidationException;
@Override
public DeviceTransportType getType() {

11
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View File

@ -362,7 +362,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
ctx.close();
} catch (AdaptorException e) {
log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
ctx.close();
sendAckOrCloseSession(ctx, topicName, msgId);
}
}
@ -451,6 +451,15 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
}
} catch (AdaptorException e) {
log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
sendAckOrCloseSession(ctx, topicName, msgId);
}
}
private void sendAckOrCloseSession(ChannelHandlerContext ctx, String topicName, int msgId) {
if (deviceSessionCtx.isSendAckOnValidationException() && msgId > 0) {
log.debug("[{}] Send pub ack on invalid publish msg [{}][{}]", sessionId, topicName, msgId);
ctx.writeAndFlush(createMqttPubAckMsg(msgId));
} else {
log.info("[{}] Closing current session due to invalid publish msg [{}][{}]", sessionId, topicName, msgId);
ctx.close();
}

7
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java
View File

@ -84,6 +84,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
private volatile MqttTransportAdaptor adaptor;
private volatile boolean jsonPayloadFormatCompatibilityEnabled;
private volatile boolean useJsonPayloadFormatForDefaultDownlinkTopics;
private volatile boolean sendAckOnValidationException;
@Getter
@Setter
@ -115,6 +116,10 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
return payloadType.equals(TransportPayloadType.JSON);
}
public boolean isSendAckOnValidationException() {
return sendAckOnValidationException;
}
public Descriptors.Descriptor getTelemetryDynamicMsgDescriptor() {
return telemetryDynamicMessageDescriptor;
}
@ -152,6 +157,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
payloadType = transportPayloadTypeConfiguration.getTransportPayloadType();
telemetryTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceTelemetryTopic());
attributesTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesTopic());
sendAckOnValidationException = mqttConfig.isSendAckOnValidationException();
if (TransportPayloadType.PROTOBUF.equals(payloadType)) {
ProtoTransportPayloadConfiguration protoTransportPayloadConfig = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
updateDynamicMessageDescriptors(protoTransportPayloadConfig);
@ -162,6 +168,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
telemetryTopicFilter = MqttTopicFilterFactory.getDefaultTelemetryFilter();
attributesTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();
payloadType = TransportPayloadType.JSON;
sendAckOnValidationException = false;
}
updateAdaptor();
}

6
ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html
View File

@ -134,4 +134,10 @@
</div>
</fieldset>
</section>
<div style="padding-bottom: 20px">
<mat-checkbox formControlName="sendAckOnValidationException">
{{ 'device-profile.mqtt-send-ack-on-validation-exception' | translate }}
</mat-checkbox>
<div class="tb-hint" innerHTML="{{ 'device-profile.mqtt-send-ack-on-validation-exception-hint' | translate }}"></div>
</div>
</form>

1
ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts
View File

@ -92,6 +92,7 @@ export class MqttDeviceProfileTransportConfigurationComponent implements Control
this.mqttDeviceProfileTransportConfigurationFormGroup = this.fb.group({
deviceAttributesTopic: [null, [Validators.required, this.validationMQTTTopic()]],
deviceTelemetryTopic: [null, [Validators.required, this.validationMQTTTopic()]],
sendAckOnValidationException: [false, Validators.required],
transportPayloadTypeConfiguration: this.fb.group({
transportPayloadType: [TransportPayloadType.JSON, Validators.required],
deviceTelemetryProtoSchema: [defaultTelemetrySchema, Validators.required],

2
ui-ngx/src/app/shared/models/device.models.ts
View File

@ -242,6 +242,7 @@ export interface DefaultDeviceProfileTransportConfiguration {
export interface MqttDeviceProfileTransportConfiguration {
deviceTelemetryTopic?: string;
deviceAttributesTopic?: string;
sendAckOnValidationException?: boolean;
transportPayloadTypeConfiguration?: {
transportPayloadType?: TransportPayloadType;
enableCompatibilityWithJsonPayloadFormat?: boolean;
@ -358,6 +359,7 @@ export function createDeviceProfileTransportConfiguration(type: DeviceTransportT
const mqttTransportConfiguration: MqttDeviceProfileTransportConfiguration = {
deviceTelemetryTopic: 'v1/devices/me/telemetry',
deviceAttributesTopic: 'v1/devices/me/attributes',
sendAckOnValidationException: false,
transportPayloadTypeConfiguration: {
transportPayloadType: TransportPayloadType.JSON,
enableCompatibilityWithJsonPayloadFormat: false,

2
ui-ngx/src/assets/locale/locale.constant-en_US.json
View File

@ -1148,6 +1148,8 @@
"mqtt-enable-compatibility-with-json-payload-format-hint": "When enabled, the platform will use a Protobuf payload format by default. If parsing fails, the platform will attempt to use JSON payload format. Useful for backward compatibility during firmware updates. For example, the initial release of the firmware uses Json, while the new release uses Protobuf. During the process of firmware update for the fleet of devices, it is required to support both Protobuf and JSON simultaneously. The compatibility mode introduces slight performance degradation, so it is recommended to disable this mode once all devices are updated.",
"mqtt-use-json-format-for-default-downlink-topics": "Use Json format for default downlink topics",
"mqtt-use-json-format-for-default-downlink-topics-hint": "When enabled, the platform will use Json payload format to push attributes and RPC via the following topics: <b>v1/devices/me/attributes/response/$request_id</b>, <b>v1/devices/me/attributes</b>, <b>v1/devices/me/rpc/request/$request_id</b>, <b>v1/devices/me/rpc/response/$request_id</b>. This setting does not impact attribute and rpc subscriptions sent using new (v2) topics: <b>v2/a/res/$request_id</b>, <b>v2/a</b>, <b>v2/r/req/$request_id</b>, <b>v2/r/res/$request_id</b>. Where <b>$request_id</b> is an integer request identifier.",
"mqtt-send-ack-on-validation-exception": "Send PUBACK on PUBLISH message validation failure",
"mqtt-send-ack-on-validation-exception-hint": "By default, the platform will close the MQTT session on message validation failure. When enabled, the platform will send publish acknowledgment instead of closing the session.",
"snmp-add-mapping": "Add SNMP mapping",
"snmp-mapping-not-configured": "No mapping for OID to timeseries/telemetry configured",
"snmp-timseries-or-attribute-name": "Timeseries/attribute name for mapping",

Write Preview
Loading…
Cancel
Save