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

Moved sending topic to MQTT transport only, parameter renamed to mqttTopic in metadata

pull/9082/head
imbeacon 3 years ago
parent
355a92588a
commit
fac2e013e2
3 changed files with 10 additions and 4 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
  2. 8
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
  3. 4
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java

2
common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View File

@ -124,7 +124,7 @@ public class DataConstants {
public static final String LAST_CONNECTED_GATEWAY = "lastConnectedGateway"; public static final String LAST_CONNECTED_GATEWAY = "lastConnectedGateway";
public static final String TOPIC = "topic"; public static final String MQTT_TOPIC = "mqttTopic";
public static final String MAIN_QUEUE_NAME = "Main"; public static final String MAIN_QUEUE_NAME = "Main";
public static final String MAIN_QUEUE_TOPIC = "tb_rule_engine.main"; public static final String MAIN_QUEUE_TOPIC = "tb_rule_engine.main";

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

@ -30,7 +30,6 @@ import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader; import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttPubAckMessage; import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage; import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubAckPayload; import io.netty.handler.codec.mqtt.MqttSubAckPayload;
@ -442,7 +441,10 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
try { try {
Matcher fwMatcher; Matcher fwMatcher;
MqttTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor(); MqttTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor();
TbMsgMetaData md = createMetadataWithTopic(topicName); TbMsgMetaData md = null;
if (deviceSessionCtx.isMqttTransportType()) {
md = createMetadataWithTopic(topicName);
}
if (deviceSessionCtx.isDeviceAttributesTopic(topicName)) { if (deviceSessionCtx.isDeviceAttributesTopic(topicName)) {
TransportProtos.PostAttributeMsg postAttributeMsg = payloadAdaptor.convertToPostAttributes(deviceSessionCtx, mqttMsg); TransportProtos.PostAttributeMsg postAttributeMsg = payloadAdaptor.convertToPostAttributes(deviceSessionCtx, mqttMsg);
transportService.process(deviceSessionCtx.getSessionInfo(), postAttributeMsg, md, getPubAckCallback(ctx, msgId, postAttributeMsg)); transportService.process(deviceSessionCtx.getSessionInfo(), postAttributeMsg, md, getPubAckCallback(ctx, msgId, postAttributeMsg));
@ -530,7 +532,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
private static TbMsgMetaData createMetadataWithTopic(String topicName) { private static TbMsgMetaData createMetadataWithTopic(String topicName) {
TbMsgMetaData md = new TbMsgMetaData(); TbMsgMetaData md = new TbMsgMetaData();
md.putValue(DataConstants.TOPIC, topicName); md.putValue(DataConstants.MQTT_TOPIC, topicName);
return md; return md;
} }

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

@ -83,6 +83,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
private volatile MqttTopicFilter attributesPublishTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter(); private volatile MqttTopicFilter attributesPublishTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();
private volatile MqttTopicFilter attributesSubscribeTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter(); private volatile MqttTopicFilter attributesSubscribeTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();
private volatile TransportPayloadType payloadType = TransportPayloadType.JSON; private volatile TransportPayloadType payloadType = TransportPayloadType.JSON;
private volatile DeviceTransportType transportType = DeviceTransportType.DEFAULT;
private volatile Descriptors.Descriptor attributesDynamicMessageDescriptor; private volatile Descriptors.Descriptor attributesDynamicMessageDescriptor;
private volatile Descriptors.Descriptor telemetryDynamicMessageDescriptor; private volatile Descriptors.Descriptor telemetryDynamicMessageDescriptor;
private volatile Descriptors.Descriptor rpcResponseDynamicMessageDescriptor; private volatile Descriptors.Descriptor rpcResponseDynamicMessageDescriptor;
@ -126,6 +127,8 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
return payloadType.equals(TransportPayloadType.JSON); return payloadType.equals(TransportPayloadType.JSON);
} }
public boolean isMqttTransportType() { return DeviceTransportType.MQTT.equals(transportType); }
public boolean isSendAckOnValidationException() { public boolean isSendAckOnValidationException() {
return sendAckOnValidationException; return sendAckOnValidationException;
} }
@ -165,6 +168,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
MqttDeviceProfileTransportConfiguration mqttConfig = (MqttDeviceProfileTransportConfiguration) transportConfiguration; MqttDeviceProfileTransportConfiguration mqttConfig = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttConfig.getTransportPayloadTypeConfiguration(); TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttConfig.getTransportPayloadTypeConfiguration();
payloadType = transportPayloadTypeConfiguration.getTransportPayloadType(); payloadType = transportPayloadTypeConfiguration.getTransportPayloadType();
transportType = DeviceTransportType.MQTT;
telemetryTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceTelemetryTopic()); telemetryTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceTelemetryTopic());
attributesPublishTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesTopic()); attributesPublishTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesTopic());
attributesSubscribeTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesSubscribeTopic()); attributesSubscribeTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesSubscribeTopic());

Write Preview
Loading…
Cancel
Save