简柏林 / thingskit · Commits

Commit 2ec09e08ff377d84b02e1223009bb76dc0aad594

Authored by zbeacon
1 parent 616f6a8b

Added ProvisionTransportType to get protocol

Inline Side-by-side
Showing 2 changed files with 11 additions and 1 deletions
@@ -42,6 +42,7 @@ import lombok.extern.slf4j.Slf4j; @@ -42,6 +42,7 @@ import lombok.extern.slf4j.Slf4j;
42 import org.thingsboard.server.common.data.DataConstants; 42 import org.thingsboard.server.common.data.DataConstants;
43 import org.thingsboard.server.common.data.DeviceProfile; 43 import org.thingsboard.server.common.data.DeviceProfile;
44 import org.thingsboard.server.common.data.DeviceTransportType; 44 import org.thingsboard.server.common.data.DeviceTransportType;
  45 +import org.thingsboard.server.common.data.TransportPayloadType;
45 import org.thingsboard.server.common.data.device.profile.MqttTopics; 46 import org.thingsboard.server.common.data.device.profile.MqttTopics;
46 import org.thingsboard.server.common.msg.EncryptionUtil; 47 import org.thingsboard.server.common.msg.EncryptionUtil;
47 import org.thingsboard.server.common.transport.SessionMsgListener; 48 import org.thingsboard.server.common.transport.SessionMsgListener;
@@ -159,6 +160,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -159,6 +160,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
159 if (e.getCause().toString().contains("JsonSyntaxException")) { 160 if (e.getCause().toString().contains("JsonSyntaxException")) {
160 TransportProtos.ProvisionDeviceRequestMsg provisionRequestMsg = deviceSessionCtx.getContext().getProtoMqttAdaptor().convertToProvisionRequestMsg(deviceSessionCtx, mqttMsg); 161 TransportProtos.ProvisionDeviceRequestMsg provisionRequestMsg = deviceSessionCtx.getContext().getProtoMqttAdaptor().convertToProvisionRequestMsg(deviceSessionCtx, mqttMsg);
161 transportService.process(provisionRequestMsg, new DeviceProvisionCallback(ctx, msgId, provisionRequestMsg)); 162 transportService.process(provisionRequestMsg, new DeviceProvisionCallback(ctx, msgId, provisionRequestMsg));
  163 + deviceSessionCtx.setProvisionPayloadType(TransportPayloadType.PROTOBUF);
162 log.trace("[{}][{}] Processing provision publish msg [{}][{}]!", sessionId, deviceSessionCtx.getDeviceId(), topicName, msgId); 164 log.trace("[{}][{}] Processing provision publish msg [{}][{}]!", sessionId, deviceSessionCtx.getDeviceId(), topicName, msgId);
163 } else { 165 } else {
164 throw e; 166 throw e;
@@ -326,7 +328,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -326,7 +328,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
326 ctx.writeAndFlush(createMqttPubAckMsg(msgId)); 328 ctx.writeAndFlush(createMqttPubAckMsg(msgId));
327 } 329 }
328 try { 330 try {
329 - deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush); 331 + if (deviceSessionCtx.getProvisionPayloadType().equals(TransportPayloadType.JSON)) {
  332 + deviceSessionCtx.getContext().getJsonMqttAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
  333 + } else {
  334 + deviceSessionCtx.getContext().getProtoMqttAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
  335 + }
330 } catch (Exception e) { 336 } catch (Exception e) {
331 log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e); 337 log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
332 } 338 }
@@ -55,6 +55,10 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext { @@ -55,6 +55,10 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
55 private volatile MqttTopicFilter attributesTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter(); 55 private volatile MqttTopicFilter attributesTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();
56 private volatile TransportPayloadType payloadType = TransportPayloadType.JSON; 56 private volatile TransportPayloadType payloadType = TransportPayloadType.JSON;
57 57
  58 + @Getter
  59 + @Setter
  60 + private TransportPayloadType provisionPayloadType = payloadType;
  61 +
58 public DeviceSessionCtx(UUID sessionId, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap, MqttTransportContext context) { 62 public DeviceSessionCtx(UUID sessionId, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap, MqttTransportContext context) {
59 super(sessionId, mqttQoSMap); 63 super(sessionId, mqttQoSMap);
60 this.context = context; 64 this.context = context;