|
|
|
@ -263,12 +263,12 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement |
|
|
|
private void processDevicePublish(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg, String topicName, int msgId) { |
|
|
|
try { |
|
|
|
MqttTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor(); |
|
|
|
if (deviceSessionCtx.isDeviceTelemetryTopic(topicName)) { |
|
|
|
TransportProtos.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, mqttMsg); |
|
|
|
transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, postTelemetryMsg)); |
|
|
|
} else if (deviceSessionCtx.isDeviceAttributesTopic(topicName)) { |
|
|
|
if (deviceSessionCtx.isDeviceAttributesTopic(topicName)) { |
|
|
|
TransportProtos.PostAttributeMsg postAttributeMsg = payloadAdaptor.convertToPostAttributes(deviceSessionCtx, mqttMsg); |
|
|
|
transportService.process(deviceSessionCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, postAttributeMsg)); |
|
|
|
} else if (deviceSessionCtx.isDeviceTelemetryTopic(topicName)) { |
|
|
|
TransportProtos.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, mqttMsg); |
|
|
|
transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, postTelemetryMsg)); |
|
|
|
} else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX)) { |
|
|
|
TransportProtos.GetAttributeRequestMsg getAttributeMsg = payloadAdaptor.convertToGetAttributes(deviceSessionCtx, mqttMsg); |
|
|
|
transportService.process(deviceSessionCtx.getSessionInfo(), getAttributeMsg, getPubAckCallback(ctx, msgId, getAttributeMsg)); |
|
|
|
|