Browse Source

added fixes after review

pull/6240/head
ShvaykaD 5 years ago
parent
commit
d4e6c013d2
  1. 4
      application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
  2. 4
      application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java
  3. 16
      application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java
  4. 8
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java
  5. 12
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java
  6. 2
      common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java
  7. 12
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
  8. 10
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java
  9. 6
      ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html
  10. 2
      ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts
  11. 4
      ui-ngx/src/app/shared/models/device.models.ts
  12. 4
      ui-ngx/src/assets/locale/locale.constant-en_US.json

4
application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java

@ -377,11 +377,11 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
return deviceProfile; return deviceProfile;
} }
protected MqttDeviceProfileTransportConfiguration createMqttDeviceProfileTransportConfiguration(TransportPayloadTypeConfiguration transportPayloadTypeConfiguration, boolean sendPubAckOnValidationException) { protected MqttDeviceProfileTransportConfiguration createMqttDeviceProfileTransportConfiguration(TransportPayloadTypeConfiguration transportPayloadTypeConfiguration, boolean sendAckOnValidationException) {
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = new MqttDeviceProfileTransportConfiguration(); MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = new MqttDeviceProfileTransportConfiguration();
mqttDeviceProfileTransportConfiguration.setDeviceTelemetryTopic(MqttTopics.DEVICE_TELEMETRY_TOPIC); mqttDeviceProfileTransportConfiguration.setDeviceTelemetryTopic(MqttTopics.DEVICE_TELEMETRY_TOPIC);
mqttDeviceProfileTransportConfiguration.setDeviceTelemetryTopic(MqttTopics.DEVICE_ATTRIBUTES_TOPIC); mqttDeviceProfileTransportConfiguration.setDeviceTelemetryTopic(MqttTopics.DEVICE_ATTRIBUTES_TOPIC);
mqttDeviceProfileTransportConfiguration.setSendPubAckOnValidationException(sendPubAckOnValidationException); mqttDeviceProfileTransportConfiguration.setSendAckOnValidationException(sendAckOnValidationException);
mqttDeviceProfileTransportConfiguration.setTransportPayloadTypeConfiguration(transportPayloadTypeConfiguration); mqttDeviceProfileTransportConfiguration.setTransportPayloadTypeConfiguration(transportPayloadTypeConfiguration);
return mqttDeviceProfileTransportConfiguration; return mqttDeviceProfileTransportConfiguration;
} }

4
application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java

@ -837,7 +837,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
} }
@Test @Test
public void testSaveDeviceProfileWithSendPubAckOnValidationException() throws Exception { public void testSaveDeviceProfileWithSendAckOnValidationException() throws Exception {
JsonTransportPayloadConfiguration jsonTransportPayloadConfiguration = new JsonTransportPayloadConfiguration(); JsonTransportPayloadConfiguration jsonTransportPayloadConfiguration = new JsonTransportPayloadConfiguration();
MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(jsonTransportPayloadConfiguration, true); MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = this.createMqttDeviceProfileTransportConfiguration(jsonTransportPayloadConfiguration, true);
DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", mqttDeviceProfileTransportConfiguration); DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile", mqttDeviceProfileTransportConfiguration);
@ -846,7 +846,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController
Assert.assertEquals(savedDeviceProfile.getTransportType(), DeviceTransportType.MQTT); Assert.assertEquals(savedDeviceProfile.getTransportType(), DeviceTransportType.MQTT);
Assert.assertTrue(savedDeviceProfile.getProfileData().getTransportConfiguration() instanceof MqttDeviceProfileTransportConfiguration); Assert.assertTrue(savedDeviceProfile.getProfileData().getTransportConfiguration() instanceof MqttDeviceProfileTransportConfiguration);
MqttDeviceProfileTransportConfiguration transportConfiguration = (MqttDeviceProfileTransportConfiguration) savedDeviceProfile.getProfileData().getTransportConfiguration(); MqttDeviceProfileTransportConfiguration transportConfiguration = (MqttDeviceProfileTransportConfiguration) savedDeviceProfile.getProfileData().getTransportConfiguration();
Assert.assertTrue(transportConfiguration.isSendPubAckOnValidationException()); Assert.assertTrue(transportConfiguration.isSendAckOnValidationException());
DeviceProfile foundDeviceProfile = doGet("/api/deviceProfile/"+ savedDeviceProfile.getId().getId().toString(), DeviceProfile.class); DeviceProfile foundDeviceProfile = doGet("/api/deviceProfile/"+ savedDeviceProfile.getId().getId().toString(), DeviceProfile.class);
Assert.assertEquals(savedDeviceProfile, foundDeviceProfile); Assert.assertEquals(savedDeviceProfile, foundDeviceProfile);
} }

16
application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java

@ -69,12 +69,12 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, enableCompatibilityWithJsonPayloadFormat, useJsonPayloadFormatForDefaultDownlinkTopics, false); 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 sendPubAckOnValidationException) throws Exception { 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, sendPubAckOnValidationException); 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 sendPubAckOnValidationException) throws Exception { 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, sendPubAckOnValidationException); this.processBeforeTest(deviceName, gatewayName, payloadType, telemetryTopic, attributesTopic, null, null, null, null, null, null, DeviceProfileProvisionType.DISABLED, false, false, sendAckOnValidationException);
} }
protected void processBeforeTest(String deviceName, protected void processBeforeTest(String deviceName,
@ -91,7 +91,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
DeviceProfileProvisionType provisionType, DeviceProfileProvisionType provisionType,
boolean enableCompatibilityWithJsonPayloadFormat, boolean enableCompatibilityWithJsonPayloadFormat,
boolean useJsonPayloadFormatForDefaultDownlinkTopics, boolean useJsonPayloadFormatForDefaultDownlinkTopics,
boolean sendPubAckOnValidationException) throws Exception { boolean sendAckOnValidationException) throws Exception {
loginSysAdmin(); loginSysAdmin();
Tenant tenant = new Tenant(); Tenant tenant = new Tenant();
@ -123,7 +123,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic, DeviceProfile mqttDeviceProfile = createMqttDeviceProfile(payloadType, telemetryTopic, attributesTopic,
telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema, telemetryProtoSchema, attributesProtoSchema, rpcResponseProtoSchema, rpcRequestProtoSchema,
provisionKey, provisionSecret, provisionType, enableCompatibilityWithJsonPayloadFormat, provisionKey, provisionSecret, provisionType, enableCompatibilityWithJsonPayloadFormat,
useJsonPayloadFormatForDefaultDownlinkTopics, sendPubAckOnValidationException); useJsonPayloadFormatForDefaultDownlinkTopics, sendAckOnValidationException);
deviceProfile = doPost("/api/deviceProfile", mqttDeviceProfile, DeviceProfile.class); deviceProfile = doPost("/api/deviceProfile", mqttDeviceProfile, DeviceProfile.class);
device.setType(deviceProfile.getName()); device.setType(deviceProfile.getName());
device.setDeviceProfileId(deviceProfile.getId()); device.setDeviceProfileId(deviceProfile.getId());
@ -182,7 +182,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
DeviceProfileProvisionType provisionType, DeviceProfileProvisionType provisionType,
boolean enableCompatibilityWithJsonPayloadFormat, boolean enableCompatibilityWithJsonPayloadFormat,
boolean useJsonPayloadFormatForDefaultDownlinkTopics, boolean useJsonPayloadFormatForDefaultDownlinkTopics,
boolean sendPubAckOnValidationException) { boolean sendAckOnValidationException) {
DeviceProfile deviceProfile = new DeviceProfile(); DeviceProfile deviceProfile = new DeviceProfile();
deviceProfile.setName(transportPayloadType.name()); deviceProfile.setName(transportPayloadType.name());
deviceProfile.setType(DeviceProfileType.DEFAULT); deviceProfile.setType(DeviceProfileType.DEFAULT);
@ -199,7 +199,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
if (!StringUtils.isEmpty(attributesTopic)) { if (!StringUtils.isEmpty(attributesTopic)) {
mqttDeviceProfileTransportConfiguration.setDeviceAttributesTopic(attributesTopic); mqttDeviceProfileTransportConfiguration.setDeviceAttributesTopic(attributesTopic);
} }
mqttDeviceProfileTransportConfiguration.setSendPubAckOnValidationException(sendPubAckOnValidationException); mqttDeviceProfileTransportConfiguration.setSendAckOnValidationException(sendAckOnValidationException);
TransportPayloadTypeConfiguration transportPayloadTypeConfiguration; TransportPayloadTypeConfiguration transportPayloadTypeConfiguration;
if (TransportPayloadType.JSON.equals(transportPayloadType)) { if (TransportPayloadType.JSON.equals(transportPayloadType)) {
transportPayloadTypeConfiguration = new JsonTransportPayloadConfiguration(); transportPayloadTypeConfiguration = new JsonTransportPayloadConfiguration();

8
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java

@ -85,7 +85,7 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
} }
@Test @Test
public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorEnabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken); MqttAsyncClient client = getMqttAsyncClient(accessToken);
@ -97,7 +97,7 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
} }
@Test @Test
public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorDisabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken); MqttAsyncClient client = getMqttAsyncClient(accessToken);
@ -109,7 +109,7 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
} }
@Test @Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendPubAckOnErrorEnabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken); MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
@ -121,7 +121,7 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
} }
@Test @Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendPubAckOnErrorDisabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken); MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);

12
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java

@ -286,7 +286,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
@Test @Test
public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorEnabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken); MqttAsyncClient client = getMqttAsyncClient(accessToken);
@ -298,7 +298,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
} }
@Test @Test
public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorDisabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken); MqttAsyncClient client = getMqttAsyncClient(accessToken);
@ -310,7 +310,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
} }
@Test @Test
public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorEnabledAndBackwardCompatibilityEnabled() throws Exception { public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorEnabledAndBackwardCompatibilityEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, true); processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, true);
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken); MqttAsyncClient client = getMqttAsyncClient(accessToken);
@ -322,7 +322,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
} }
@Test @Test
public void testPushTelemetryWithMalformedPayloadAndSendPubAckOnErrorDisabledAndBackwardCompatibilityEnabled() throws Exception { public void testPushTelemetryWithMalformedPayloadAndSendAckOnErrorDisabledAndBackwardCompatibilityEnabled() throws Exception {
processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, false); processBeforeTest("Test Post Telemetry device", "Test Post Telemetry gateway", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null, true, false, false);
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(accessToken); MqttAsyncClient client = getMqttAsyncClient(accessToken);
@ -334,7 +334,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
} }
@Test @Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendPubAckOnErrorEnabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken); MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
@ -346,7 +346,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
} }
@Test @Test
public void testPushTelemetryGatewayWithMalformedPayloadAndSendPubAckOnErrorDisabled() throws Exception { 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); 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); CountDownLatch latch = new CountDownLatch(1);
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken); MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);

2
common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java

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

12
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

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

10
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java

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

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

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

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

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

4
ui-ngx/src/app/shared/models/device.models.ts

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

4
ui-ngx/src/assets/locale/locale.constant-en_US.json

@ -1135,8 +1135,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-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": "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-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-pub-ack-on-validation-exception": "Send PUBACK on PUBLISH message validation failure", "mqtt-send-ack-on-validation-exception": "Send PUBACK on PUBLISH message validation failure",
"mqtt-send-pub-ack-on-validation-exception-hint": "When enabled, the MQTT transport service will send publish acknowledgment on publish message validation failure, otherwise, the MQTT transport service will close the MQTT session.", "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-add-mapping": "Add SNMP mapping",
"snmp-mapping-not-configured": "No mapping for OID to timeseries/telemetry configured", "snmp-mapping-not-configured": "No mapping for OID to timeseries/telemetry configured",
"snmp-timseries-or-attribute-name": "Timeseries/attribute name for mapping", "snmp-timseries-or-attribute-name": "Timeseries/attribute name for mapping",

Loading…
Cancel
Save