简柏林 / thingskit · Commits

Commit 845d8247dd185a4dbae3cc43a6fa10ae3b033ea7

Authored by ShvaykaD
Committed by GitHub
1 parent 08c6a80d

[3.3.2] MQTT short topics support (#4967) 

* added mqtt short topics support

* remove volatile keyword from topic types parameters

* added new tests for mqtt short topics

* fix compilation error after merge

* improvements/typo-fixes after pull request review
Inline Side-by-side
Showing 23 changed files with 709 additions and 260 deletions
... ... @@ -52,7 +52,17 @@ public abstract class AbstractMqttAttributesRequestIntegrationTest extends Abstr
52 52
53 53 @Test
54 54 public void testRequestAttributesValuesFromTheServer() throws Exception {
55   - processTestRequestAttributesValuesFromTheServer();
  55 + processTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
  56 + }
  57 +
  58 + @Test
  59 + public void testRequestAttributesValuesFromTheServerOnShortTopic() throws Exception {
  60 + processTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX);
  61 + }
  62 +
  63 + @Test
  64 + public void testRequestAttributesValuesFromTheServerOnShortJsonTopic() throws Exception {
  65 + processTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_JSON_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_JSON_TOPIC_PREFIX);
56 66 }
57 67
58 68 @Test
... ... @@ -60,18 +70,18 @@ public abstract class AbstractMqttAttributesRequestIntegrationTest extends Abstr
60 70 processTestGatewayRequestAttributesValuesFromTheServer();
61 71 }
62 72
63   - protected void processTestRequestAttributesValuesFromTheServer() throws Exception {
  73 + protected void processTestRequestAttributesValuesFromTheServer(String attrPubTopic, String attrSubTopic, String attrReqTopicPrefix) throws Exception {
64 74
65 75 MqttAsyncClient client = getMqttAsyncClient(accessToken);
66 76
67   - postAttributesAndSubscribeToTopic(savedDevice, client);
  77 + postAttributesAndSubscribeToTopic(savedDevice, client, attrPubTopic, attrSubTopic);
68 78
69 79 Thread.sleep(5000);
70 80
71 81 TestMqttCallback callback = getTestMqttCallback();
72 82 client.setCallback(callback);
73 83
74   - validateResponse(client, callback.getLatch(), callback);
  84 + validateResponse(client, callback.getLatch(), callback, attrReqTopicPrefix);
75 85 }
76 86
77 87 protected void processTestGatewayRequestAttributesValuesFromTheServer() throws Exception {
... ... @@ -103,10 +113,10 @@ public abstract class AbstractMqttAttributesRequestIntegrationTest extends Abstr
103 113 validateSharedResponseGateway(client, sharedAttributesCallback);
104 114 }
105 115
106   - protected void postAttributesAndSubscribeToTopic(Device savedDevice, MqttAsyncClient client) throws Exception {
  116 + protected void postAttributesAndSubscribeToTopic(Device savedDevice, MqttAsyncClient client, String attrPubTopic, String attrSubTopic) throws Exception {
107 117 doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", POST_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
108   - client.publish(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, new MqttMessage(POST_ATTRIBUTES_PAYLOAD.getBytes())).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
109   - client.subscribe(MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttQoS.AT_MOST_ONCE.value()).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
  118 + client.publish(attrPubTopic, new MqttMessage(POST_ATTRIBUTES_PAYLOAD.getBytes())).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
  119 + client.subscribe(attrSubTopic, MqttQoS.AT_MOST_ONCE.value()).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
110 120 }
111 121
112 122 protected void postGatewayDeviceClientAttributes(MqttAsyncClient client) throws Exception {
... ... @@ -114,12 +124,12 @@ public abstract class AbstractMqttAttributesRequestIntegrationTest extends Abstr
114 124 client.publish(MqttTopics.GATEWAY_ATTRIBUTES_TOPIC, new MqttMessage(postClientAttributes.getBytes())).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
115 125 }
116 126
117   - protected void validateResponse(MqttAsyncClient client, CountDownLatch latch, TestMqttCallback callback) throws MqttException, InterruptedException, InvalidProtocolBufferException {
  127 + protected void validateResponse(MqttAsyncClient client, CountDownLatch latch, TestMqttCallback callback, String attrReqTopicPrefix) throws MqttException, InterruptedException, InvalidProtocolBufferException {
118 128 String keys = "attribute1,attribute2,attribute3,attribute4,attribute5";
119 129 String payloadStr = "{\"clientKeys\":\"" + keys + "\", \"sharedKeys\":\"" + keys + "\"}";
120 130 MqttMessage mqttMessage = new MqttMessage();
121 131 mqttMessage.setPayload(payloadStr.getBytes());
122   - client.publish(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX + "1", mqttMessage).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
  132 + client.publish(attrReqTopicPrefix + "1", mqttMessage).waitForCompletion(TimeUnit.MINUTES.toMillis(1));
123 133 latch.await(1, TimeUnit.MINUTES);
124 134 assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
125 135 String expectedRequestPayload = "{\"client\":{\"attribute1\":\"value1\",\"attribute2\":true,\"attribute3\":42.0,\"attribute4\":73,\"attribute5\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}},\"shared\":{\"attribute1\":\"value1\",\"attribute2\":true,\"attribute3\":42.0,\"attribute4\":73,\"attribute5\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}}";
... ...
... ... @@ -20,6 +20,7 @@ import org.junit.After;
20 20 import org.junit.Before;
21 21 import org.junit.Test;
22 22 import org.thingsboard.server.common.data.TransportPayloadType;
  23 +import org.thingsboard.server.common.data.device.profile.MqttTopics;
23 24
24 25 @Slf4j
25 26 public abstract class AbstractMqttAttributesRequestJsonIntegrationTest extends AbstractMqttAttributesRequestIntegrationTest {
... ... @@ -36,7 +37,17 @@ public abstract class AbstractMqttAttributesRequestJsonIntegrationTest extends A
36 37
37 38 @Test
38 39 public void testRequestAttributesValuesFromTheServer() throws Exception {
39   - processTestRequestAttributesValuesFromTheServer();
  40 + super.testRequestAttributesValuesFromTheServer();
  41 + }
  42 +
  43 + @Test
  44 + public void testRequestAttributesValuesFromTheServerOnShortTopic() throws Exception {
  45 + super.testRequestAttributesValuesFromTheServerOnShortTopic();
  46 + }
  47 +
  48 + @Test
  49 + public void testRequestAttributesValuesFromTheServerOnShortJsonTopic() throws Exception {
  50 + super.testRequestAttributesValuesFromTheServerOnShortJsonTopic();
40 51 }
41 52
42 53 @Test
... ...
... ... @@ -84,7 +84,21 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends
84 84 public void testRequestAttributesValuesFromTheServer() throws Exception {
85 85 super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto",
86 86 TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED);
87   - processTestRequestAttributesValuesFromTheServer();
  87 + processTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
  88 + }
  89 +
  90 + @Test
  91 + public void testRequestAttributesValuesFromTheServerOnShortTopic() throws Exception {
  92 + super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto",
  93 + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED);
  94 + processTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX);
  95 + }
  96 +
  97 + @Test
  98 + public void testRequestAttributesValuesFromTheServerOnShortProtoTopic() throws Exception {
  99 + super.processBeforeTest("Test Request attribute values from the server proto", "Gateway Test Request attribute values from the server proto",
  100 + TransportPayloadType.PROTOBUF, null, null, null, ATTRIBUTES_SCHEMA_STR, null, null, null, null, DeviceProfileProvisionType.DISABLED);
  101 + processTestRequestAttributesValuesFromTheServer(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX);
88 102 }
89 103
90 104 @Test
... ... @@ -93,7 +107,10 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends
93 107 processTestGatewayRequestAttributesValuesFromTheServer();
94 108 }
95 109
96   - protected void postAttributesAndSubscribeToTopic(Device savedDevice, MqttAsyncClient client) throws Exception {
  110 + @Test
  111 + public void testRequestAttributesValuesFromTheServerOnShortJsonTopic() throws Exception { }
  112 +
  113 + protected void postAttributesAndSubscribeToTopic(Device savedDevice, MqttAsyncClient client, String attrPubTopic, String attrSubTopic) throws Exception {
97 114 doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", AbstractMqttAttributesIntegrationTest.POST_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
98 115 DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
99 116 assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
... ... @@ -131,8 +148,8 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends
131 148 .setField(postAttributesMsgDescriptor.findFieldByName("attribute5"), jsonObject)
132 149 .build();
133 150 byte[] payload = postAttributesMsg.toByteArray();
134   - client.publish(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, new MqttMessage(payload));
135   - client.subscribe(MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttQoS.AT_MOST_ONCE.value());
  151 + client.publish(attrPubTopic, new MqttMessage(payload));
  152 + client.subscribe(attrSubTopic, MqttQoS.AT_MOST_ONCE.value());
136 153 }
137 154
138 155 protected void postGatewayDeviceClientAttributes(MqttAsyncClient client) throws Exception {
... ... @@ -149,7 +166,7 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends
149 166 client.publish(MqttTopics.GATEWAY_ATTRIBUTES_TOPIC, new MqttMessage(bytes));
150 167 }
151 168
152   - protected void validateResponse(MqttAsyncClient client, CountDownLatch latch, AbstractMqttAttributesIntegrationTest.TestMqttCallback callback) throws MqttException, InterruptedException, InvalidProtocolBufferException {
  169 + protected void validateResponse(MqttAsyncClient client, CountDownLatch latch, TestMqttCallback callback, String attrReqTopic) throws MqttException, InterruptedException, InvalidProtocolBufferException {
153 170 String keys = "attribute1,attribute2,attribute3,attribute4,attribute5";
154 171 TransportApiProtos.AttributesRequest.Builder attributesRequestBuilder = TransportApiProtos.AttributesRequest.newBuilder();
155 172 attributesRequestBuilder.setClientKeys(keys);
... ... @@ -157,7 +174,7 @@ public abstract class AbstractMqttAttributesRequestProtoIntegrationTest extends
157 174 TransportApiProtos.AttributesRequest attributesRequest = attributesRequestBuilder.build();
158 175 MqttMessage mqttMessage = new MqttMessage();
159 176 mqttMessage.setPayload(attributesRequest.toByteArray());
160   - client.publish(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX + "1", mqttMessage);
  177 + client.publish(attrReqTopic + "1", mqttMessage);
161 178 latch.await(3, TimeUnit.SECONDS);
162 179 assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
163 180 TransportProtos.GetAttributeResponseMsg expectedAttributesResponse = getExpectedAttributeResponseMsg();
... ...
... ... @@ -61,7 +61,17 @@ public abstract class AbstractMqttAttributesUpdatesIntegrationTest extends Abstr
61 61
62 62 @Test
63 63 public void testSubscribeToAttributesUpdatesFromTheServer() throws Exception {
64   - processTestSubscribeToAttributesUpdates();
  64 + processTestSubscribeToAttributesUpdates(MqttTopics.DEVICE_ATTRIBUTES_TOPIC);
  65 + }
  66 +
  67 + @Test
  68 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortTopic() throws Exception {
  69 + processTestSubscribeToAttributesUpdates(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC);
  70 + }
  71 +
  72 + @Test
  73 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortJsonTopic() throws Exception {
  74 + processTestSubscribeToAttributesUpdates(MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC);
65 75 }
66 76
67 77 @Test
... ... @@ -69,14 +79,14 @@ public abstract class AbstractMqttAttributesUpdatesIntegrationTest extends Abstr
69 79 processGatewayTestSubscribeToAttributesUpdates();
70 80 }
71 81
72   - protected void processTestSubscribeToAttributesUpdates() throws Exception {
  82 + protected void processTestSubscribeToAttributesUpdates(String attrSubTopic) throws Exception {
73 83
74 84 MqttAsyncClient client = getMqttAsyncClient(accessToken);
75 85
76 86 TestMqttCallback onUpdateCallback = getTestMqttCallback();
77 87 client.setCallback(onUpdateCallback);
78 88
79   - client.subscribe(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE.value());
  89 + client.subscribe(attrSubTopic, MqttQoS.AT_MOST_ONCE.value());
80 90
81 91 Thread.sleep(1000);
82 92
... ...
... ... @@ -36,7 +36,17 @@ public abstract class AbstractMqttAttributesUpdatesJsonIntegrationTest extends A
36 36
37 37 @Test
38 38 public void testSubscribeToAttributesUpdatesFromTheServer() throws Exception {
39   - processTestSubscribeToAttributesUpdates();
  39 + super.testSubscribeToAttributesUpdatesFromTheServer();
  40 + }
  41 +
  42 + @Test
  43 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortTopic() throws Exception {
  44 + super.testSubscribeToAttributesUpdatesFromTheServerOnShortTopic();
  45 + }
  46 +
  47 + @Test
  48 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortJsonTopic() throws Exception {
  49 + super.testSubscribeToAttributesUpdatesFromTheServerOnShortJsonTopic();
40 50 }
41 51
42 52 @Test
... ...
... ... @@ -21,6 +21,7 @@ import org.junit.After;
21 21 import org.junit.Before;
22 22 import org.junit.Test;
23 23 import org.thingsboard.server.common.data.TransportPayloadType;
  24 +import org.thingsboard.server.common.data.device.profile.MqttTopics;
24 25 import org.thingsboard.server.gen.transport.TransportApiProtos;
25 26 import org.thingsboard.server.gen.transport.TransportProtos;
26 27
... ... @@ -46,7 +47,20 @@ public abstract class AbstractMqttAttributesUpdatesProtoIntegrationTest extends
46 47
47 48 @Test
48 49 public void testSubscribeToAttributesUpdatesFromTheServer() throws Exception {
49   - processTestSubscribeToAttributesUpdates();
  50 + super.testSubscribeToAttributesUpdatesFromTheServer();
  51 + }
  52 +
  53 + @Test
  54 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortTopic() throws Exception {
  55 + super.testSubscribeToAttributesUpdatesFromTheServerOnShortTopic();
  56 + }
  57 +
  58 + @Test
  59 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortJsonTopic() throws Exception {}
  60 +
  61 + @Test
  62 + public void testSubscribeToAttributesUpdatesFromTheServerOnShortProtoTopic() throws Exception {
  63 + processTestSubscribeToAttributesUpdates(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC);
50 64 }
51 65
52 66 @Test
... ...
... ... @@ -21,6 +21,7 @@ import org.junit.After;
21 21 import org.junit.Assert;
22 22 import org.junit.Before;
23 23 import org.junit.Test;
  24 +import org.thingsboard.server.common.data.device.profile.MqttTopics;
24 25 import org.thingsboard.server.service.security.AccessValidator;
25 26
26 27 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
... ... @@ -81,12 +82,32 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
81 82
82 83 @Test
83 84 public void testServerMqttOneWayRpc() throws Exception {
84   - processOneWayRpcTest();
  85 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
  86 + }
  87 +
  88 + @Test
  89 + public void testServerMqttOneWayRpcOnShortTopic() throws Exception {
  90 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
  91 + }
  92 +
  93 + @Test
  94 + public void testServerMqttOneWayRpcOnShortJsonTopic() throws Exception {
  95 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC);
85 96 }
86 97
87 98 @Test
88 99 public void testServerMqttTwoWayRpc() throws Exception {
89   - processTwoWayRpcTest();
  100 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
  101 + }
  102 +
  103 + @Test
  104 + public void testServerMqttTwoWayRpcOnShortTopic() throws Exception {
  105 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
  106 + }
  107 +
  108 + @Test
  109 + public void testServerMqttTwoWayRpcOnShortJsonTopic() throws Exception {
  110 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC);
90 111 }
91 112
92 113 @Test
... ...
... ... @@ -60,14 +60,14 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
60 60 asyncContextTimeoutToUseRpcPlugin = 10000L;
61 61 }
62 62
63   - protected void processOneWayRpcTest() throws Exception {
  63 + protected void processOneWayRpcTest(String rpcSubTopic) throws Exception {
64 64 MqttAsyncClient client = getMqttAsyncClient(accessToken);
65 65
66 66 CountDownLatch latch = new CountDownLatch(1);
67 67 TestMqttCallback callback = new TestMqttCallback(client, latch);
68 68 client.setCallback(callback);
69 69
70   - client.subscribe(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC, MqttQoS.AT_MOST_ONCE.value());
  70 + client.subscribe(rpcSubTopic, MqttQoS.AT_MOST_ONCE.value());
71 71
72 72 Thread.sleep(1000);
73 73
... ... @@ -86,9 +86,9 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
86 86 validateOneWayRpcGatewayResponse(deviceName, client, payloadBytes);
87 87 }
88 88
89   - protected void processTwoWayRpcTest() throws Exception {
  89 + protected void processTwoWayRpcTest(String rpcSubTopic) throws Exception {
90 90 MqttAsyncClient client = getMqttAsyncClient(accessToken);
91   - client.subscribe(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC, 1);
  91 + client.subscribe(rpcSubTopic, 1);
92 92
93 93 CountDownLatch latch = new CountDownLatch(1);
94 94 TestMqttCallback callback = new TestMqttCallback(client, latch);
... ... @@ -199,7 +199,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
199 199
200 200 protected MqttMessage processMessageArrived(String requestTopic, MqttMessage mqttMessage) throws MqttException, InvalidProtocolBufferException {
201 201 MqttMessage message = new MqttMessage();
202   - if (requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC)) {
  202 + if (requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC) || requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC_V2)) {
203 203 message.setPayload(DEVICE_RESPONSE.getBytes(StandardCharset.UTF_8));
204 204 } else {
205 205 JsonNode requestMsgNode = JacksonUtil.toJsonNode(new String(mqttMessage.getPayload(), StandardCharset.UTF_8));
... ... @@ -232,7 +232,12 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
232 232 @Override
233 233 public void messageArrived(String requestTopic, MqttMessage mqttMessage) throws Exception {
234 234 log.info("Message Arrived: " + Arrays.toString(mqttMessage.getPayload()));
235   - String responseTopic = requestTopic.replace("request", "response");
  235 + String responseTopic;
  236 + if (requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC_V2)) {
  237 + responseTopic = requestTopic.replace("req", "res");
  238 + } else {
  239 + responseTopic = requestTopic.replace("request", "response");
  240 + }
236 241 qoS = mqttMessage.getQos();
237 242 client.publish(responseTopic, processMessageArrived(requestTopic, mqttMessage));
238 243 latch.countDown();
... ...
... ... @@ -21,6 +21,7 @@ import org.junit.After;
21 21 import org.junit.Before;
22 22 import org.junit.Test;
23 23 import org.thingsboard.server.common.data.TransportPayloadType;
  24 +import org.thingsboard.server.common.data.device.profile.MqttTopics;
24 25
25 26 @Slf4j
26 27 public abstract class AbstractMqttServerSideRpcJsonIntegrationTest extends AbstractMqttServerSideRpcIntegrationTest {
... ... @@ -37,12 +38,32 @@ public abstract class AbstractMqttServerSideRpcJsonIntegrationTest extends Abstr
37 38
38 39 @Test
39 40 public void testServerMqttOneWayRpc() throws Exception {
40   - processOneWayRpcTest();
  41 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
  42 + }
  43 +
  44 + @Test
  45 + public void testServerMqttOneWayRpcOnShortTopic() throws Exception {
  46 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
  47 + }
  48 +
  49 + @Test
  50 + public void testServerMqttOneWayRpcOnShortJsonTopic() throws Exception {
  51 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC);
41 52 }
42 53
43 54 @Test
44 55 public void testServerMqttTwoWayRpc() throws Exception {
45   - processTwoWayRpcTest();
  56 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
  57 + }
  58 +
  59 + @Test
  60 + public void testServerMqttTwoWayRpcOnShortTopic() throws Exception {
  61 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
  62 + }
  63 +
  64 + @Test
  65 + public void testServerMqttTwoWayRpcOnShortJsonTopic() throws Exception {
  66 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC);
46 67 }
47 68
48 69 @Test
... ...
... ... @@ -78,12 +78,32 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst
78 78
79 79 @Test
80 80 public void testServerMqttOneWayRpc() throws Exception {
81   - processOneWayRpcTest();
  81 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
  82 + }
  83 +
  84 + @Test
  85 + public void testServerMqttOneWayRpcOnShortTopic() throws Exception {
  86 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
  87 + }
  88 +
  89 + @Test
  90 + public void testServerMqttOneWayRpcOnShortProtoTopic() throws Exception {
  91 + processOneWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC);
82 92 }
83 93
84 94 @Test
85 95 public void testServerMqttTwoWayRpc() throws Exception {
86   - processTwoWayRpcTest();
  96 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC);
  97 + }
  98 +
  99 + @Test
  100 + public void testServerMqttTwoWayRpcOnShortTopic() throws Exception {
  101 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC);
  102 + }
  103 +
  104 + @Test
  105 + public void testServerMqttTwoWayRpcOnShortProtoTopic() throws Exception {
  106 + processTwoWayRpcTest(MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC);
87 107 }
88 108
89 109 @Test
... ... @@ -118,9 +138,9 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst
118 138 return builder.build();
119 139 }
120 140
121   - protected void processTwoWayRpcTest() throws Exception {
  141 + protected void processTwoWayRpcTest(String rpcSubTopic) throws Exception {
122 142 MqttAsyncClient client = getMqttAsyncClient(accessToken);
123   - client.subscribe(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC, 1);
  143 + client.subscribe(rpcSubTopic, 1);
124 144
125 145 CountDownLatch latch = new CountDownLatch(1);
126 146 TestMqttCallback callback = new TestMqttCallback(client, latch);
... ... @@ -139,7 +159,7 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst
139 159
140 160 protected MqttMessage processMessageArrived(String requestTopic, MqttMessage mqttMessage) throws MqttException, InvalidProtocolBufferException {
141 161 MqttMessage message = new MqttMessage();
142   - if (requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC)) {
  162 + if (requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC) || requestTopic.startsWith(MqttTopics.BASE_DEVICE_API_TOPIC_V2)) {
143 163 ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = getProtoTransportPayloadConfiguration();
144 164 ProtoFileElement rpcRequestProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(RPC_REQUEST_PROTO_SCHEMA);
145 165 DynamicSchema rpcRequestProtoSchema = protoTransportPayloadConfiguration.getDynamicSchema(rpcRequestProtoSchemaFile, ProtoTransportPayloadConfiguration.RPC_REQUEST_PROTO_SCHEMA);
... ...
... ... @@ -61,6 +61,18 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
61 61 }
62 62
63 63 @Test
  64 + public void testPushAttributesOnShortTopic() throws Exception {
  65 + List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
  66 + processJsonPayloadAttributesTest(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, expectedKeys, PAYLOAD_VALUES_STR.getBytes());
  67 + }
  68 +
  69 + @Test
  70 + public void testPushAttributesOnShortJsonTopic() throws Exception {
  71 + List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
  72 + processJsonPayloadAttributesTest(MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC, expectedKeys, PAYLOAD_VALUES_STR.getBytes());
  73 + }
  74 +
  75 + @Test
64 76 public void testPushAttributesGateway() throws Exception {
65 77 List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
66 78 String deviceName1 = "Device A";
... ...
... ... @@ -20,6 +20,7 @@ import org.junit.After;
20 20 import org.junit.Before;
21 21 import org.junit.Test;
22 22 import org.thingsboard.server.common.data.TransportPayloadType;
  23 +import org.thingsboard.server.common.data.device.profile.MqttTopics;
23 24
24 25 import java.util.Arrays;
25 26 import java.util.List;
... ... @@ -46,11 +47,17 @@ public abstract class AbstractMqttAttributesJsonIntegrationTest extends Abstract
46 47 }
47 48
48 49 @Test
  50 + public void testPushAttributesOnShortTopic() throws Exception {
  51 + super.testPushAttributesOnShortTopic();
  52 + }
  53 +
  54 + @Test
  55 + public void testPushAttributesOnShortJsonTopic() throws Exception {
  56 + super.testPushAttributesOnShortJsonTopic();
  57 + }
  58 +
  59 + @Test
49 60 public void testPushAttributesGateway() throws Exception {
50   - List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
51   - String deviceName1 = "Device A";
52   - String deviceName2 = "Device B";
53   - String payload = getGatewayAttributesJsonPayload(deviceName1, deviceName2);
54   - processGatewayAttributesTest(expectedKeys, payload.getBytes(), deviceName1, deviceName2);
  61 + super.testPushAttributesGateway();
55 62 }
56 63 }
... ...
... ... @@ -25,6 +25,7 @@ import org.junit.Test;
25 25 import org.thingsboard.server.common.data.TransportPayloadType;
26 26 import org.thingsboard.server.common.data.device.profile.DeviceProfileTransportConfiguration;
27 27 import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration;
  28 +import org.thingsboard.server.common.data.device.profile.MqttTopics;
28 29 import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration;
29 30 import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration;
30 31 import org.thingsboard.server.gen.transport.TransportApiProtos;
... ... @@ -49,14 +50,14 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
49 50 @Test
50 51 public void testPushAttributes() throws Exception {
51 52 super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, POST_DATA_ATTRIBUTES_TOPIC);
52   - DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
53   - assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
54   - MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
55   - TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration();
56   - assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
57   - ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
58   - ProtoFileElement transportProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(DEVICE_ATTRIBUTES_PROTO_SCHEMA);
59   - DynamicSchema attributesSchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchemaFile, ProtoTransportPayloadConfiguration.ATTRIBUTES_PROTO_SCHEMA);
  53 + DynamicMessage postAttributesMsg = getDefaultDynamicMessage();
  54 + processAttributesTest(POST_DATA_ATTRIBUTES_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), false);
  55 + }
  56 +
  57 + @Test
  58 + public void testPushAttributesWithExplicitPresenceProtoKeys() throws Exception {
  59 + super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, POST_DATA_ATTRIBUTES_TOPIC);
  60 + DynamicSchema attributesSchema = getDynamicSchema();
60 61
61 62 DynamicMessage.Builder nestedJsonObjectBuilder = attributesSchema.newMessageBuilder("PostAttributes.JsonObject.NestedJsonObject");
62 63 Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
... ... @@ -67,7 +68,6 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
67 68 Descriptors.Descriptor jsonObjectBuilderDescriptor = jsonObjectBuilder.getDescriptorForType();
68 69 assertNotNull(jsonObjectBuilderDescriptor);
69 70 DynamicMessage jsonObject = jsonObjectBuilder
70   - .setField(jsonObjectBuilderDescriptor.findFieldByName("someNumber"), 42)
71 71 .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 1)
72 72 .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 2)
73 73 .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 3)
... ... @@ -78,18 +78,50 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
78 78 Descriptors.Descriptor postAttributesMsgDescriptor = postAttributesBuilder.getDescriptorForType();
79 79 assertNotNull(postAttributesMsgDescriptor);
80 80 DynamicMessage postAttributesMsg = postAttributesBuilder
81   - .setField(postAttributesMsgDescriptor.findFieldByName("key1"), "value1")
82   - .setField(postAttributesMsgDescriptor.findFieldByName("key2"), true)
83   - .setField(postAttributesMsgDescriptor.findFieldByName("key3"), 3.0)
84   - .setField(postAttributesMsgDescriptor.findFieldByName("key4"), 4)
  81 + .setField(postAttributesMsgDescriptor.findFieldByName("key1"), "")
  82 + .setField(postAttributesMsgDescriptor.findFieldByName("key2"), false)
  83 + .setField(postAttributesMsgDescriptor.findFieldByName("key3"), 0.0)
  84 + .setField(postAttributesMsgDescriptor.findFieldByName("key4"), 0)
85 85 .setField(postAttributesMsgDescriptor.findFieldByName("key5"), jsonObject)
86 86 .build();
87   - processAttributesTest(POST_DATA_ATTRIBUTES_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), false);
  87 + processAttributesTest(POST_DATA_ATTRIBUTES_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), true);
88 88 }
89 89
90 90 @Test
91   - public void testPushAttributesWithExplicitPresenceProtoKeys() throws Exception {
  91 + public void testPushAttributesOnShortTopic() throws Exception {
92 92 super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, POST_DATA_ATTRIBUTES_TOPIC);
  93 + DynamicMessage postAttributesMsg = getDefaultDynamicMessage();
  94 + processAttributesTest(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), false);
  95 + }
  96 +
  97 + @Test
  98 + public void testPushAttributesOnShortJsonTopic() throws Exception {
  99 + super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, POST_DATA_ATTRIBUTES_TOPIC);
  100 + processJsonPayloadAttributesTest(MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), PAYLOAD_VALUES_STR.getBytes());
  101 + }
  102 +
  103 + @Test
  104 + public void testPushAttributesOnShortProtoTopic() throws Exception {
  105 + super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, POST_DATA_ATTRIBUTES_TOPIC);
  106 + DynamicMessage postAttributesMsg = getDefaultDynamicMessage();
  107 + processAttributesTest(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), false);
  108 + }
  109 +
  110 + @Test
  111 + public void testPushAttributesGateway() throws Exception {
  112 + super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, null);
  113 + TransportApiProtos.GatewayAttributesMsg.Builder gatewayAttributesMsgProtoBuilder = TransportApiProtos.GatewayAttributesMsg.newBuilder();
  114 + List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
  115 + String deviceName1 = "Device A";
  116 + String deviceName2 = "Device B";
  117 + TransportApiProtos.AttributesMsg firstDeviceAttributesMsgProto = getDeviceAttributesMsgProto(deviceName1, expectedKeys);
  118 + TransportApiProtos.AttributesMsg secondDeviceAttributesMsgProto = getDeviceAttributesMsgProto(deviceName2, expectedKeys);
  119 + gatewayAttributesMsgProtoBuilder.addAllMsg(Arrays.asList(firstDeviceAttributesMsgProto, secondDeviceAttributesMsgProto));
  120 + TransportApiProtos.GatewayAttributesMsg gatewayAttributesMsg = gatewayAttributesMsgProtoBuilder.build();
  121 + processGatewayAttributesTest(expectedKeys, gatewayAttributesMsg.toByteArray(), deviceName1, deviceName2);
  122 + }
  123 +
  124 + private DynamicSchema getDynamicSchema() {
93 125 DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
94 126 assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
95 127 MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
... ... @@ -97,7 +129,11 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
97 129 assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
98 130 ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
99 131 ProtoFileElement transportProtoSchemaFile = protoTransportPayloadConfiguration.getTransportProtoSchema(DEVICE_ATTRIBUTES_PROTO_SCHEMA);
100   - DynamicSchema attributesSchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchemaFile, ProtoTransportPayloadConfiguration.ATTRIBUTES_PROTO_SCHEMA);
  132 + return protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchemaFile, ProtoTransportPayloadConfiguration.ATTRIBUTES_PROTO_SCHEMA);
  133 + }
  134 +
  135 + private DynamicMessage getDefaultDynamicMessage() {
  136 + DynamicSchema attributesSchema = getDynamicSchema();
101 137
102 138 DynamicMessage.Builder nestedJsonObjectBuilder = attributesSchema.newMessageBuilder("PostAttributes.JsonObject.NestedJsonObject");
103 139 Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
... ... @@ -108,6 +144,7 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
108 144 Descriptors.Descriptor jsonObjectBuilderDescriptor = jsonObjectBuilder.getDescriptorForType();
109 145 assertNotNull(jsonObjectBuilderDescriptor);
110 146 DynamicMessage jsonObject = jsonObjectBuilder
  147 + .setField(jsonObjectBuilderDescriptor.findFieldByName("someNumber"), 42)
111 148 .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 1)
112 149 .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 2)
113 150 .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 3)
... ... @@ -117,25 +154,13 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
117 154 DynamicMessage.Builder postAttributesBuilder = attributesSchema.newMessageBuilder("PostAttributes");
118 155 Descriptors.Descriptor postAttributesMsgDescriptor = postAttributesBuilder.getDescriptorForType();
119 156 assertNotNull(postAttributesMsgDescriptor);
120   - DynamicMessage postAttributesMsg = postAttributesBuilder
121   - .setField(postAttributesMsgDescriptor.findFieldByName("key1"), "")
  157 + return postAttributesBuilder
  158 + .setField(postAttributesMsgDescriptor.findFieldByName("key1"), "value1")
  159 + .setField(postAttributesMsgDescriptor.findFieldByName("key2"), true)
  160 + .setField(postAttributesMsgDescriptor.findFieldByName("key3"), 3.0)
  161 + .setField(postAttributesMsgDescriptor.findFieldByName("key4"), 4)
122 162 .setField(postAttributesMsgDescriptor.findFieldByName("key5"), jsonObject)
123 163 .build();
124   - processAttributesTest(POST_DATA_ATTRIBUTES_TOPIC, Arrays.asList("key1", "key5"), postAttributesMsg.toByteArray(), true);
125   - }
126   -
127   - @Test
128   - public void testPushAttributesGateway() throws Exception {
129   - super.processBeforeTest("Test Post Attributes device", "Test Post Attributes gateway", TransportPayloadType.PROTOBUF, null, null);
130   - TransportApiProtos.GatewayAttributesMsg.Builder gatewayAttributesMsgProtoBuilder = TransportApiProtos.GatewayAttributesMsg.newBuilder();
131   - List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
132   - String deviceName1 = "Device A";
133   - String deviceName2 = "Device B";
134   - TransportApiProtos.AttributesMsg firstDeviceAttributesMsgProto = getDeviceAttributesMsgProto(deviceName1, expectedKeys);
135   - TransportApiProtos.AttributesMsg secondDeviceAttributesMsgProto = getDeviceAttributesMsgProto(deviceName2, expectedKeys);
136   - gatewayAttributesMsgProtoBuilder.addAllMsg(Arrays.asList(firstDeviceAttributesMsgProto, secondDeviceAttributesMsgProto));
137   - TransportApiProtos.GatewayAttributesMsg gatewayAttributesMsg = gatewayAttributesMsgProtoBuilder.build();
138   - processGatewayAttributesTest(expectedKeys, gatewayAttributesMsg.toByteArray(), deviceName1, deviceName2);
139 164 }
140 165
141 166 private TransportApiProtos.AttributesMsg getDeviceAttributesMsgProto(String deviceName, List<String> expectedKeys) {
... ...
... ... @@ -74,6 +74,18 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
74 74 }
75 75
76 76 @Test
  77 + public void testPushTelemetryOnShortTopic() throws Exception {
  78 + List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
  79 + processJsonPayloadTelemetryTest(MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC, expectedKeys, PAYLOAD_VALUES_STR.getBytes(), false);
  80 + }
  81 +
  82 + @Test
  83 + public void testPushTelemetryOnShortJsonTopic() throws Exception {
  84 + List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
  85 + processJsonPayloadTelemetryTest(MqttTopics.DEVICE_TELEMETRY_SHORT_JSON_TOPIC, expectedKeys, PAYLOAD_VALUES_STR.getBytes(), false);
  86 + }
  87 +
  88 + @Test
77 89 public void testPushTelemetryGateway() throws Exception {
78 90 List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
79 91 String deviceName1 = "Device A";
... ...
... ... @@ -58,24 +58,22 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract
58 58 }
59 59
60 60 @Test
  61 + public void testPushTelemetryOnShortTopic() throws Exception {
  62 + super.testPushTelemetryOnShortTopic();
  63 + }
  64 +
  65 + @Test
  66 + public void testPushTelemetryWithTsOnShortJsonTopic() throws Exception {
  67 + super.testPushTelemetryOnShortJsonTopic();
  68 + }
  69 +
  70 + @Test
61 71 public void testPushTelemetryGateway() throws Exception {
62   - List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5");
63   - String deviceName1 = "Device A";
64   - String deviceName2 = "Device B";
65   - String payload = getGatewayTelemetryJsonPayload(deviceName1, deviceName2, "10000", "20000");
66   - processGatewayTelemetryTest(MqttTopics.GATEWAY_TELEMETRY_TOPIC, expectedKeys, payload.getBytes(), deviceName1, deviceName2);
  72 + super.testPushTelemetryGateway();
67 73 }
68 74
69 75 @Test
70 76 public void testGatewayConnect() throws Exception {
71   - String payload = "{\"device\":\"Device A\", \"type\": \"" + TransportPayloadType.JSON.name() + "\"}";
72   - MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
73   - publishMqttMsg(client, payload.getBytes(), MqttTopics.GATEWAY_CONNECT_TOPIC);
74   -
75   - String deviceName = "Device A";
76   - Device device = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class),
77   - 20,
78   - 100);
79   - assertNotNull(device);
  77 + super.testGatewayConnect();
80 78 }
81 79 }
... ...
... ... @@ -21,6 +21,7 @@ import com.google.protobuf.DynamicMessage;
21 21 import com.squareup.wire.schema.internal.parser.ProtoFileElement;
22 22 import lombok.extern.slf4j.Slf4j;
23 23 import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
  24 +import org.jetbrains.annotations.NotNull;
24 25 import org.junit.After;
25 26 import org.junit.Ignore;
26 27 import org.junit.Test;
... ... @@ -55,41 +56,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
55 56 @Test
56 57 public void testPushTelemetry() throws Exception {
57 58 super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null);
58   - DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
59   - assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
60   - MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
61   - TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration();
62   - assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
63   - ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
64   - ProtoFileElement transportProtoSchema = protoTransportPayloadConfiguration.getTransportProtoSchema(DEVICE_TELEMETRY_PROTO_SCHEMA);
65   - DynamicSchema telemetrySchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchema, "telemetrySchema");
66   -
67   - DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
68   - Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
69   - assertNotNull(nestedJsonObjectBuilderDescriptor);
70   - DynamicMessage nestedJsonObject = nestedJsonObjectBuilder.setField(nestedJsonObjectBuilderDescriptor.findFieldByName("key"), "value").build();
71   -
72   - DynamicMessage.Builder jsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject");
73   - Descriptors.Descriptor jsonObjectBuilderDescriptor = jsonObjectBuilder.getDescriptorForType();
74   - assertNotNull(jsonObjectBuilderDescriptor);
75   - DynamicMessage jsonObject = jsonObjectBuilder
76   - .setField(jsonObjectBuilderDescriptor.findFieldByName("someNumber"), 42)
77   - .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 1)
78   - .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 2)
79   - .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 3)
80   - .setField(jsonObjectBuilderDescriptor.findFieldByName("someNestedObject"), nestedJsonObject)
81   - .build();
82   -
83   - DynamicMessage.Builder postTelemetryBuilder = telemetrySchema.newMessageBuilder("PostTelemetry");
84   - Descriptors.Descriptor postTelemetryMsgDescriptor = postTelemetryBuilder.getDescriptorForType();
85   - assertNotNull(postTelemetryMsgDescriptor);
86   - DynamicMessage postTelemetryMsg = postTelemetryBuilder
87   - .setField(postTelemetryMsgDescriptor.findFieldByName("key1"), "value1")
88   - .setField(postTelemetryMsgDescriptor.findFieldByName("key2"), true)
89   - .setField(postTelemetryMsgDescriptor.findFieldByName("key3"), 3.0)
90   - .setField(postTelemetryMsgDescriptor.findFieldByName("key4"), 4)
91   - .setField(postTelemetryMsgDescriptor.findFieldByName("key5"), jsonObject)
92   - .build();
  59 + DynamicMessage postTelemetryMsg = getDefaultDynamicMessage();
93 60 processTelemetryTest(POST_DATA_TELEMETRY_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postTelemetryMsg.toByteArray(), false, false);
94 61 }
95 62
... ... @@ -121,14 +88,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
121 88 " }\n" +
122 89 "}";
123 90 super.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);
124   - DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
125   - assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
126   - MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
127   - TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration();
128   - assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
129   - ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
130   - ProtoFileElement transportProtoSchema = protoTransportPayloadConfiguration.getTransportProtoSchema(schemaStr);
131   - DynamicSchema telemetrySchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchema, "telemetrySchema");
  91 + DynamicSchema telemetrySchema = getDynamicSchema(schemaStr);
132 92
133 93 DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
134 94 Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
... ... @@ -173,14 +133,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
173 133 @Test
174 134 public void testPushTelemetryWithExplicitPresenceProtoKeys() throws Exception {
175 135 super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null);
176   - DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
177   - assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
178   - MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
179   - TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration();
180   - assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
181   - ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
182   - ProtoFileElement transportProtoSchema = protoTransportPayloadConfiguration.getTransportProtoSchema(DEVICE_TELEMETRY_PROTO_SCHEMA);
183   - DynamicSchema telemetrySchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchema, "telemetrySchema");
  136 + DynamicSchema telemetrySchema = getDynamicSchema(DEVICE_TELEMETRY_PROTO_SCHEMA);
184 137
185 138 DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
186 139 Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
... ... @@ -237,14 +190,7 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
237 190 " }\n" +
238 191 "}";
239 192 super.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);
240   - DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
241   - assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
242   - MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
243   - TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration();
244   - assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
245   - ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
246   - ProtoFileElement transportProtoSchema = protoTransportPayloadConfiguration.getTransportProtoSchema(schemaStr);
247   - DynamicSchema telemetrySchema = protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchema, "telemetrySchema");
  193 + DynamicSchema telemetrySchema = getDynamicSchema(schemaStr);
248 194
249 195 DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
250 196 Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
... ... @@ -282,6 +228,26 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
282 228 }
283 229
284 230 @Test
  231 + public void testPushTelemetryOnShortTopic() throws Exception {
  232 + super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null);
  233 + DynamicMessage postTelemetryMsg = getDefaultDynamicMessage();
  234 + processTelemetryTest(MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postTelemetryMsg.toByteArray(), false, false);
  235 + }
  236 +
  237 + @Test
  238 + public void testPushTelemetryOnShortJsonTopic() throws Exception {
  239 + super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null);
  240 + processJsonPayloadTelemetryTest(MqttTopics.DEVICE_TELEMETRY_SHORT_JSON_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), PAYLOAD_VALUES_STR.getBytes(), false);
  241 + }
  242 +
  243 + @Test
  244 + public void testPushTelemetryOnShortProtoTopic() throws Exception {
  245 + super.processBeforeTest("Test Post Telemetry device proto payload", "Test Post Telemetry gateway proto payload", TransportPayloadType.PROTOBUF, POST_DATA_TELEMETRY_TOPIC, null);
  246 + DynamicMessage postTelemetryMsg = getDefaultDynamicMessage();
  247 + processTelemetryTest(MqttTopics.DEVICE_TELEMETRY_SHORT_PROTO_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postTelemetryMsg.toByteArray(), false, false);
  248 + }
  249 +
  250 + @Test
285 251 public void testPushTelemetryGateway() throws Exception {
286 252 super.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);
287 253 TransportApiProtos.GatewayTelemetryMsg.Builder gatewayTelemetryMsgProtoBuilder = TransportApiProtos.GatewayTelemetryMsg.newBuilder();
... ... @@ -310,6 +276,48 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
310 276 assertNotNull(device);
311 277 }
312 278
  279 + private DynamicSchema getDynamicSchema(String deviceTelemetryProtoSchema) {
  280 + DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
  281 + assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration);
  282 + MqttDeviceProfileTransportConfiguration mqttTransportConfiguration = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
  283 + TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttTransportConfiguration.getTransportPayloadTypeConfiguration();
  284 + assertTrue(transportPayloadTypeConfiguration instanceof ProtoTransportPayloadConfiguration);
  285 + ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
  286 + ProtoFileElement transportProtoSchema = protoTransportPayloadConfiguration.getTransportProtoSchema(deviceTelemetryProtoSchema);
  287 + return protoTransportPayloadConfiguration.getDynamicSchema(transportProtoSchema, "telemetrySchema");
  288 + }
  289 +
  290 + private DynamicMessage getDefaultDynamicMessage() {
  291 + DynamicSchema telemetrySchema = getDynamicSchema(DEVICE_TELEMETRY_PROTO_SCHEMA);
  292 +
  293 + DynamicMessage.Builder nestedJsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject.NestedJsonObject");
  294 + Descriptors.Descriptor nestedJsonObjectBuilderDescriptor = nestedJsonObjectBuilder.getDescriptorForType();
  295 + assertNotNull(nestedJsonObjectBuilderDescriptor);
  296 + DynamicMessage nestedJsonObject = nestedJsonObjectBuilder.setField(nestedJsonObjectBuilderDescriptor.findFieldByName("key"), "value").build();
  297 +
  298 + DynamicMessage.Builder jsonObjectBuilder = telemetrySchema.newMessageBuilder("PostTelemetry.JsonObject");
  299 + Descriptors.Descriptor jsonObjectBuilderDescriptor = jsonObjectBuilder.getDescriptorForType();
  300 + assertNotNull(jsonObjectBuilderDescriptor);
  301 + DynamicMessage jsonObject = jsonObjectBuilder
  302 + .setField(jsonObjectBuilderDescriptor.findFieldByName("someNumber"), 42)
  303 + .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 1)
  304 + .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 2)
  305 + .addRepeatedField(jsonObjectBuilderDescriptor.findFieldByName("someArray"), 3)
  306 + .setField(jsonObjectBuilderDescriptor.findFieldByName("someNestedObject"), nestedJsonObject)
  307 + .build();
  308 +
  309 + DynamicMessage.Builder postTelemetryBuilder = telemetrySchema.newMessageBuilder("PostTelemetry");
  310 + Descriptors.Descriptor postTelemetryMsgDescriptor = postTelemetryBuilder.getDescriptorForType();
  311 + assertNotNull(postTelemetryMsgDescriptor);
  312 + return postTelemetryBuilder
  313 + .setField(postTelemetryMsgDescriptor.findFieldByName("key1"), "value1")
  314 + .setField(postTelemetryMsgDescriptor.findFieldByName("key2"), true)
  315 + .setField(postTelemetryMsgDescriptor.findFieldByName("key3"), 3.0)
  316 + .setField(postTelemetryMsgDescriptor.findFieldByName("key4"), 4)
  317 + .setField(postTelemetryMsgDescriptor.findFieldByName("key5"), jsonObject)
  318 + .build();
  319 + }
  320 +
313 321 private TransportApiProtos.ConnectMsg getConnectProto(String deviceName) {
314 322 TransportApiProtos.ConnectMsg.Builder builder = TransportApiProtos.ConnectMsg.newBuilder();
315 323 builder.setDeviceName(deviceName);
... ...
... ... @@ -34,20 +34,25 @@ public class MqttTopics {
34 34 private static final String SOFTWARE = "/sw";
35 35 private static final String CHUNK = "/chunk/";
36 36 private static final String ERROR = "/error";
37   -
  37 + private static final String TELEMETRY_SHORT = "/t";
  38 + private static final String ATTRIBUTES_SHORT = "/a";
  39 + private static final String RPC_SHORT = "/r";
  40 + private static final String REQUEST_SHORT = "/req";
  41 + private static final String RESPONSE_SHORT = "/res";
  42 + private static final String JSON_SHORT = "j";
  43 + private static final String PROTO_SHORT = "p";
38 44 private static final String ATTRIBUTES_RESPONSE = ATTRIBUTES + RESPONSE;
39 45 private static final String ATTRIBUTES_REQUEST = ATTRIBUTES + REQUEST;
40   -
  46 + private static final String ATTRIBUTES_RESPONSE_SHORT = ATTRIBUTES_SHORT + RESPONSE_SHORT + "/";
  47 + private static final String ATTRIBUTES_REQUEST_SHORT = ATTRIBUTES_SHORT + REQUEST_SHORT + "/";
41 48 private static final String DEVICE_RPC_RESPONSE = RPC + RESPONSE + "/";
42 49 private static final String DEVICE_RPC_REQUEST = RPC + REQUEST + "/";
43   -
  50 + private static final String DEVICE_RPC_RESPONSE_SHORT = RPC_SHORT + RESPONSE_SHORT + "/";
  51 + private static final String DEVICE_RPC_REQUEST_SHORT = RPC_SHORT + REQUEST_SHORT + "/";
44 52 private static final String DEVICE_ATTRIBUTES_RESPONSE = ATTRIBUTES_RESPONSE + "/";
45 53 private static final String DEVICE_ATTRIBUTES_REQUEST = ATTRIBUTES_REQUEST + "/";
46   -
47   - // V1_JSON topics
48   -
  54 + // v1 topics
49 55 public static final String BASE_DEVICE_API_TOPIC = "v1/devices/me";
50   -
51 56 public static final String DEVICE_RPC_RESPONSE_TOPIC = BASE_DEVICE_API_TOPIC + DEVICE_RPC_RESPONSE;
52 57 public static final String DEVICE_RPC_RESPONSE_SUB_TOPIC = DEVICE_RPC_RESPONSE_TOPIC + SUB_TOPIC;
53 58 public static final String DEVICE_RPC_REQUESTS_TOPIC = BASE_DEVICE_API_TOPIC + DEVICE_RPC_REQUEST;
... ... @@ -60,9 +65,7 @@ public class MqttTopics {
60 65 public static final String DEVICE_ATTRIBUTES_TOPIC = BASE_DEVICE_API_TOPIC + ATTRIBUTES;
61 66 public static final String DEVICE_PROVISION_REQUEST_TOPIC = PROVISION + REQUEST;
62 67 public static final String DEVICE_PROVISION_RESPONSE_TOPIC = PROVISION + RESPONSE;
63   -
64   - // V1_JSON gateway topics
65   -
  68 + // v1 gateway topics
66 69 public static final String BASE_GATEWAY_API_TOPIC = "v1/gateway";
67 70 public static final String GATEWAY_CONNECT_TOPIC = BASE_GATEWAY_API_TOPIC + CONNECT;
68 71 public static final String GATEWAY_DISCONNECT_TOPIC = BASE_GATEWAY_API_TOPIC + DISCONNECT;
... ... @@ -72,22 +75,44 @@ public class MqttTopics {
72 75 public static final String GATEWAY_RPC_TOPIC = BASE_GATEWAY_API_TOPIC + RPC;
73 76 public static final String GATEWAY_ATTRIBUTES_REQUEST_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_REQUEST;
74 77 public static final String GATEWAY_ATTRIBUTES_RESPONSE_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_RESPONSE;
75   -
76 78 // v2 topics
77 79 public static final String BASE_DEVICE_API_TOPIC_V2 = "v2";
78   -
79 80 public static final String REQUEST_ID_PATTERN = "(?<requestId>\\d+)";
80 81 public static final String CHUNK_PATTERN = "(?<chunk>\\d+)";
81   -
82 82 public static final String DEVICE_FIRMWARE_REQUEST_TOPIC_PATTERN = BASE_DEVICE_API_TOPIC_V2 + FIRMWARE + REQUEST + "/" + REQUEST_ID_PATTERN + CHUNK + CHUNK_PATTERN;
83 83 public static final String DEVICE_FIRMWARE_RESPONSES_TOPIC = BASE_DEVICE_API_TOPIC_V2 + FIRMWARE + RESPONSE + "/" + SUB_TOPIC + CHUNK + SUB_TOPIC;
84 84 public static final String DEVICE_FIRMWARE_ERROR_TOPIC = BASE_DEVICE_API_TOPIC_V2 + FIRMWARE + ERROR;
85   -
86 85 public static final String DEVICE_SOFTWARE_FIRMWARE_RESPONSES_TOPIC_FORMAT = BASE_DEVICE_API_TOPIC_V2 + "/%s" + RESPONSE + "/%s" + CHUNK + "%d";
87   -
88 86 public static final String DEVICE_SOFTWARE_REQUEST_TOPIC_PATTERN = BASE_DEVICE_API_TOPIC_V2 + SOFTWARE + REQUEST + "/" + REQUEST_ID_PATTERN + CHUNK + CHUNK_PATTERN;
89 87 public static final String DEVICE_SOFTWARE_RESPONSES_TOPIC = BASE_DEVICE_API_TOPIC_V2 + SOFTWARE + RESPONSE + "/" + SUB_TOPIC + CHUNK + SUB_TOPIC;
90 88 public static final String DEVICE_SOFTWARE_ERROR_TOPIC = BASE_DEVICE_API_TOPIC_V2 + SOFTWARE + ERROR;
  89 + public static final String DEVICE_ATTRIBUTES_SHORT_TOPIC = BASE_DEVICE_API_TOPIC_V2 + ATTRIBUTES_SHORT;
  90 + public static final String DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC = BASE_DEVICE_API_TOPIC_V2 + ATTRIBUTES_SHORT + "/" + JSON_SHORT;
  91 + public static final String DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC = BASE_DEVICE_API_TOPIC_V2 + ATTRIBUTES_SHORT + "/" + PROTO_SHORT;
  92 + public static final String DEVICE_TELEMETRY_SHORT_TOPIC = BASE_DEVICE_API_TOPIC_V2 + TELEMETRY_SHORT;
  93 + public static final String DEVICE_TELEMETRY_SHORT_JSON_TOPIC = BASE_DEVICE_API_TOPIC_V2 + TELEMETRY_SHORT + "/" + JSON_SHORT;
  94 + public static final String DEVICE_TELEMETRY_SHORT_PROTO_TOPIC = BASE_DEVICE_API_TOPIC_V2 + TELEMETRY_SHORT + "/" + PROTO_SHORT;
  95 + public static final String DEVICE_RPC_RESPONSE_SHORT_TOPIC = BASE_DEVICE_API_TOPIC_V2 + DEVICE_RPC_RESPONSE_SHORT;
  96 + public static final String DEVICE_RPC_RESPONSE_SHORT_JSON_TOPIC = BASE_DEVICE_API_TOPIC_V2 + DEVICE_RPC_RESPONSE_SHORT + JSON_SHORT + "/";
  97 + public static final String DEVICE_RPC_RESPONSE_SHORT_PROTO_TOPIC = BASE_DEVICE_API_TOPIC_V2 + DEVICE_RPC_RESPONSE_SHORT + PROTO_SHORT + "/";
  98 + public static final String DEVICE_RPC_RESPONSE_SUB_SHORT_TOPIC = DEVICE_RPC_RESPONSE_SHORT_TOPIC + SUB_TOPIC;
  99 + public static final String DEVICE_RPC_RESPONSE_SUB_SHORT_JSON_TOPIC = DEVICE_RPC_RESPONSE_SHORT_TOPIC + JSON_SHORT + "/" + SUB_TOPIC;
  100 + public static final String DEVICE_RPC_RESPONSE_SUB_SHORT_PROTO_TOPIC = DEVICE_RPC_RESPONSE_SHORT_TOPIC + PROTO_SHORT + "/" + SUB_TOPIC;
  101 + public static final String DEVICE_RPC_REQUESTS_SHORT_TOPIC = BASE_DEVICE_API_TOPIC_V2 + DEVICE_RPC_REQUEST_SHORT;
  102 + public static final String DEVICE_RPC_REQUESTS_SHORT_JSON_TOPIC = BASE_DEVICE_API_TOPIC_V2 + DEVICE_RPC_REQUEST_SHORT + JSON_SHORT + "/";
  103 + public static final String DEVICE_RPC_REQUESTS_SHORT_PROTO_TOPIC = BASE_DEVICE_API_TOPIC_V2 + DEVICE_RPC_REQUEST_SHORT + PROTO_SHORT + "/";
  104 + public static final String DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC = DEVICE_RPC_REQUESTS_SHORT_TOPIC + SUB_TOPIC;
  105 + public static final String DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC = DEVICE_RPC_REQUESTS_SHORT_TOPIC + JSON_SHORT + "/" + SUB_TOPIC;
  106 + public static final String DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC = DEVICE_RPC_REQUESTS_SHORT_TOPIC + PROTO_SHORT + "/" + SUB_TOPIC;
  107 + public static final String DEVICE_ATTRIBUTES_RESPONSE_SHORT_TOPIC_PREFIX = BASE_DEVICE_API_TOPIC_V2 + ATTRIBUTES_RESPONSE_SHORT;
  108 + public static final String DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC = DEVICE_ATTRIBUTES_RESPONSE_SHORT_TOPIC_PREFIX + SUB_TOPIC;
  109 + public static final String DEVICE_ATTRIBUTES_RESPONSE_SHORT_JSON_TOPIC_PREFIX = DEVICE_ATTRIBUTES_RESPONSE_SHORT_TOPIC_PREFIX + JSON_SHORT + "/";
  110 + public static final String DEVICE_ATTRIBUTES_RESPONSES_SHORT_JSON_TOPIC = DEVICE_ATTRIBUTES_RESPONSE_SHORT_JSON_TOPIC_PREFIX + SUB_TOPIC;
  111 + public static final String DEVICE_ATTRIBUTES_RESPONSE_SHORT_PROTO_TOPIC_PREFIX = DEVICE_ATTRIBUTES_RESPONSE_SHORT_TOPIC_PREFIX + PROTO_SHORT + "/";
  112 + public static final String DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC = DEVICE_ATTRIBUTES_RESPONSE_SHORT_PROTO_TOPIC_PREFIX + SUB_TOPIC;
  113 + public static final String DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX = BASE_DEVICE_API_TOPIC_V2 + ATTRIBUTES_REQUEST_SHORT;
  114 + public static final String DEVICE_ATTRIBUTES_REQUEST_SHORT_JSON_TOPIC_PREFIX = DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX + JSON_SHORT + "/";
  115 + public static final String DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX = DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX + PROTO_SHORT + "/";
91 116
92 117 private MqttTopics() {
93 118 }
... ...
... ... @@ -20,7 +20,6 @@ import com.google.gson.JsonParser;
20 20 import com.google.protobuf.Descriptors;
21 21 import com.google.protobuf.DynamicMessage;
22 22 import com.google.protobuf.InvalidProtocolBufferException;
23   -import com.google.protobuf.util.JsonFormat;
24 23 import lombok.extern.slf4j.Slf4j;
25 24 import org.eclipse.californium.core.coap.CoAP;
26 25 import org.eclipse.californium.core.coap.MediaTypeRegistry;
... ... @@ -32,6 +31,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
32 31 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
33 32 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
34 33 import org.thingsboard.server.common.transport.adaptor.ProtoConverter;
  34 +import org.thingsboard.server.gen.transport.TransportApiProtos;
35 35 import org.thingsboard.server.gen.transport.TransportProtos;
36 36 import org.thingsboard.server.transport.coap.CoapTransportResource;
37 37
... ... @@ -44,8 +44,9 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor {
44 44
45 45 @Override
46 46 public TransportProtos.PostTelemetryMsg convertToPostTelemetry(UUID sessionId, Request inbound, Descriptors.Descriptor telemetryMsgDescriptor) throws AdaptorException {
  47 + ProtoConverter.validateDescriptor(telemetryMsgDescriptor);
47 48 try {
48   - return JsonConverter.convertToTelemetryProto(new JsonParser().parse(dynamicMsgToJson(inbound.getPayload(), telemetryMsgDescriptor)));
  49 + return JsonConverter.convertToTelemetryProto(new JsonParser().parse(ProtoConverter.dynamicMsgToJson(inbound.getPayload(), telemetryMsgDescriptor)));
49 50 } catch (Exception e) {
50 51 throw new AdaptorException(e);
51 52 }
... ... @@ -53,8 +54,9 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor {
53 54
54 55 @Override
55 56 public TransportProtos.PostAttributeMsg convertToPostAttributes(UUID sessionId, Request inbound, Descriptors.Descriptor attributesMsgDescriptor) throws AdaptorException {
  57 + ProtoConverter.validateDescriptor(attributesMsgDescriptor);
56 58 try {
57   - return JsonConverter.convertToAttributesProto(new JsonParser().parse(dynamicMsgToJson(inbound.getPayload(), attributesMsgDescriptor)));
  59 + return JsonConverter.convertToAttributesProto(new JsonParser().parse(ProtoConverter.dynamicMsgToJson(inbound.getPayload(), attributesMsgDescriptor)));
58 60 } catch (Exception e) {
59 61 throw new AdaptorException(e);
60 62 }
... ... @@ -71,8 +73,9 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor {
71 73 if (requestId.isEmpty()) {
72 74 throw new AdaptorException("Request id is missing!");
73 75 } else {
  76 + ProtoConverter.validateDescriptor(rpcResponseMsgDescriptor);
74 77 try {
75   - JsonElement response = new JsonParser().parse(dynamicMsgToJson(inbound.getPayload(), rpcResponseMsgDescriptor));
  78 + JsonElement response = new JsonParser().parse(ProtoConverter.dynamicMsgToJson(inbound.getPayload(), rpcResponseMsgDescriptor));
76 79 return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId.orElseThrow(() -> new AdaptorException("Request id is missing!")))
77 80 .setPayload(response.toString()).build();
78 81 } catch (Exception e) {
... ... @@ -158,11 +161,6 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor {
158 161 return response;
159 162 }
160 163
161   - private String dynamicMsgToJson(byte[] bytes, Descriptors.Descriptor descriptor) throws InvalidProtocolBufferException {
162   - DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor, bytes);
163   - return JsonFormat.printer().includingDefaultValueFields().print(dynamicMessage);
164   - }
165   -
166 164 @Override
167 165 public int getContentFormat() {
168 166 return MediaTypeRegistry.APPLICATION_OCTET_STREAM;
... ...
... ... @@ -101,8 +101,6 @@ import static io.netty.handler.codec.mqtt.MqttMessageType.UNSUBACK;
101 101 import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;
102 102 import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
103 103 import static io.netty.handler.codec.mqtt.MqttQoS.FAILURE;
104   -import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_FIRMWARE_REQUEST_TOPIC_PATTERN;
105   -import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_SOFTWARE_REQUEST_TOPIC_PATTERN;
106 104
107 105 /**
108 106 * @author Andrew Shvayka
... ... @@ -110,8 +108,8 @@ import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVIC
110 108 @Slf4j
111 109 public class MqttTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener<Future<? super Void>>, SessionMsgListener {
112 110
113   - private static final Pattern FW_REQUEST_PATTERN = Pattern.compile(DEVICE_FIRMWARE_REQUEST_TOPIC_PATTERN);
114   - private static final Pattern SW_REQUEST_PATTERN = Pattern.compile(DEVICE_SOFTWARE_REQUEST_TOPIC_PATTERN);
  111 + private static final Pattern FW_REQUEST_PATTERN = Pattern.compile(MqttTopics.DEVICE_FIRMWARE_REQUEST_TOPIC_PATTERN);
  112 + private static final Pattern SW_REQUEST_PATTERN = Pattern.compile(MqttTopics.DEVICE_SOFTWARE_REQUEST_TOPIC_PATTERN);
115 113
116 114
117 115 private static final String PAYLOAD_TOO_LARGE = "PAYLOAD_TOO_LARGE";
... ... @@ -133,6 +131,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
133 131 private final ConcurrentHashMap<String, Integer> chunkSizes;
134 132 private final ConcurrentMap<Integer, TransportProtos.ToDeviceRpcRequestMsg> rpcAwaitingAck;
135 133
  134 + private TopicType attrSubTopicType;
  135 + private TopicType rpcSubTopicType;
  136 + private TopicType attrReqTopicType;
  137 + private TopicType toServerRpcSubTopicType;
  138 +
136 139 MqttTransportHandler(MqttTransportContext context, SslHandler sslHandler) {
137 140 this.sessionId = UUID.randomUUID();
138 141 this.context = context;
... ... @@ -355,14 +358,16 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
355 358 TransportProtos.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, mqttMsg);
356 359 transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, postTelemetryMsg));
357 360 } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX)) {
358   - TransportProtos.GetAttributeRequestMsg getAttributeMsg = payloadAdaptor.convertToGetAttributes(deviceSessionCtx, mqttMsg);
  361 + TransportProtos.GetAttributeRequestMsg getAttributeMsg = payloadAdaptor.convertToGetAttributes(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
359 362 transportService.process(deviceSessionCtx.getSessionInfo(), getAttributeMsg, getPubAckCallback(ctx, msgId, getAttributeMsg));
  363 + attrReqTopicType = TopicType.V1;
360 364 } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_TOPIC)) {
361   - TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = payloadAdaptor.convertToDeviceRpcResponse(deviceSessionCtx, mqttMsg);
  365 + TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = payloadAdaptor.convertToDeviceRpcResponse(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC);
362 366 transportService.process(deviceSessionCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(ctx, msgId, rpcResponseMsg));
363 367 } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_TOPIC)) {
364   - TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = payloadAdaptor.convertToServerRpcRequest(deviceSessionCtx, mqttMsg);
  368 + TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = payloadAdaptor.convertToServerRpcRequest(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC);
365 369 transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
  370 + toServerRpcSubTopicType = TopicType.V1;
366 371 } else if (topicName.equals(MqttTopics.DEVICE_CLAIM_TOPIC)) {
367 372 TransportProtos.ClaimDeviceMsg claimDeviceMsg = payloadAdaptor.convertToClaimDevice(deviceSessionCtx, mqttMsg);
368 373 transportService.process(deviceSessionCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(ctx, msgId, claimDeviceMsg));
... ... @@ -370,6 +375,57 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
370 375 getOtaPackageCallback(ctx, mqttMsg, msgId, fwMatcher, OtaPackageType.FIRMWARE);
371 376 } else if ((fwMatcher = SW_REQUEST_PATTERN.matcher(topicName)).find()) {
372 377 getOtaPackageCallback(ctx, mqttMsg, msgId, fwMatcher, OtaPackageType.SOFTWARE);
  378 + } else if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC)) {
  379 + TransportProtos.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, mqttMsg);
  380 + transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, postTelemetryMsg));
  381 + } else if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_SHORT_JSON_TOPIC)) {
  382 + TransportProtos.PostTelemetryMsg postTelemetryMsg = context.getJsonMqttAdaptor().convertToPostTelemetry(deviceSessionCtx, mqttMsg);
  383 + transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, postTelemetryMsg));
  384 + } else if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_SHORT_PROTO_TOPIC)) {
  385 + TransportProtos.PostTelemetryMsg postTelemetryMsg = context.getProtoMqttAdaptor().convertToPostTelemetry(deviceSessionCtx, mqttMsg);
  386 + transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, postTelemetryMsg));
  387 + } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC)) {
  388 + TransportProtos.PostAttributeMsg postAttributeMsg = payloadAdaptor.convertToPostAttributes(deviceSessionCtx, mqttMsg);
  389 + transportService.process(deviceSessionCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, postAttributeMsg));
  390 + } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC)) {
  391 + TransportProtos.PostAttributeMsg postAttributeMsg = context.getJsonMqttAdaptor().convertToPostAttributes(deviceSessionCtx, mqttMsg);
  392 + transportService.process(deviceSessionCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, postAttributeMsg));
  393 + } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC)) {
  394 + TransportProtos.PostAttributeMsg postAttributeMsg = context.getProtoMqttAdaptor().convertToPostAttributes(deviceSessionCtx, mqttMsg);
  395 + transportService.process(deviceSessionCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, postAttributeMsg));
  396 + } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_SHORT_JSON_TOPIC)) {
  397 + TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = context.getJsonMqttAdaptor().convertToDeviceRpcResponse(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_RESPONSE_SHORT_JSON_TOPIC);
  398 + transportService.process(deviceSessionCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(ctx, msgId, rpcResponseMsg));
  399 + } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_SHORT_PROTO_TOPIC)) {
  400 + TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = context.getProtoMqttAdaptor().convertToDeviceRpcResponse(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_RESPONSE_SHORT_PROTO_TOPIC);
  401 + transportService.process(deviceSessionCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(ctx, msgId, rpcResponseMsg));
  402 + } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_SHORT_TOPIC)) {
  403 + TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = payloadAdaptor.convertToDeviceRpcResponse(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_RESPONSE_SHORT_TOPIC);
  404 + transportService.process(deviceSessionCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(ctx, msgId, rpcResponseMsg));
  405 + } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_SHORT_JSON_TOPIC)) {
  406 + TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = context.getJsonMqttAdaptor().convertToServerRpcRequest(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_REQUESTS_SHORT_JSON_TOPIC);
  407 + transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
  408 + toServerRpcSubTopicType = TopicType.V2_JSON;
  409 + } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_SHORT_PROTO_TOPIC)) {
  410 + TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = context.getProtoMqttAdaptor().convertToServerRpcRequest(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_REQUESTS_SHORT_PROTO_TOPIC);
  411 + transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
  412 + toServerRpcSubTopicType = TopicType.V2_PROTO;
  413 + } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_SHORT_TOPIC)) {
  414 + TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = payloadAdaptor.convertToServerRpcRequest(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_RPC_REQUESTS_SHORT_TOPIC);
  415 + transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
  416 + toServerRpcSubTopicType = TopicType.V2;
  417 + } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_JSON_TOPIC_PREFIX)) {
  418 + TransportProtos.GetAttributeRequestMsg getAttributeMsg = context.getJsonMqttAdaptor().convertToGetAttributes(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_JSON_TOPIC_PREFIX);
  419 + transportService.process(deviceSessionCtx.getSessionInfo(), getAttributeMsg, getPubAckCallback(ctx, msgId, getAttributeMsg));
  420 + attrReqTopicType = TopicType.V2_JSON;
  421 + } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX)) {
  422 + TransportProtos.GetAttributeRequestMsg getAttributeMsg = context.getProtoMqttAdaptor().convertToGetAttributes(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX);
  423 + transportService.process(deviceSessionCtx.getSessionInfo(), getAttributeMsg, getPubAckCallback(ctx, msgId, getAttributeMsg));
  424 + attrReqTopicType = TopicType.V2_PROTO;
  425 + } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX)) {
  426 + TransportProtos.GetAttributeRequestMsg getAttributeMsg = payloadAdaptor.convertToGetAttributes(deviceSessionCtx, mqttMsg, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX);
  427 + transportService.process(deviceSessionCtx.getSessionInfo(), getAttributeMsg, getPubAckCallback(ctx, msgId, getAttributeMsg));
  428 + attrReqTopicType = TopicType.V2;
373 429 } else {
374 430 transportService.reportActivity(deviceSessionCtx.getSessionInfo());
375 431 ack(ctx, msgId);
... ... @@ -541,19 +597,53 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
541 597 try {
542 598 switch (topic) {
543 599 case MqttTopics.DEVICE_ATTRIBUTES_TOPIC: {
544   - transportService.process(deviceSessionCtx.getSessionInfo(), TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null);
545   - registerSubQoS(topic, grantedQoSList, reqQoS);
  600 + processAttributesSubscribe(grantedQoSList, topic, reqQoS, TopicType.V1);
  601 + activityReported = true;
  602 + break;
  603 + }
  604 + case MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC: {
  605 + processAttributesSubscribe(grantedQoSList, topic, reqQoS, TopicType.V2);
  606 + activityReported = true;
  607 + break;
  608 + }
  609 + case MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC: {
  610 + processAttributesSubscribe(grantedQoSList, topic, reqQoS, TopicType.V2_JSON);
  611 + activityReported = true;
  612 + break;
  613 + }
  614 + case MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC: {
  615 + processAttributesSubscribe(grantedQoSList, topic, reqQoS, TopicType.V2_PROTO);
546 616 activityReported = true;
547 617 break;
548 618 }
549 619 case MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC: {
550   - transportService.process(deviceSessionCtx.getSessionInfo(), TransportProtos.SubscribeToRPCMsg.newBuilder().build(), null);
551   - registerSubQoS(topic, grantedQoSList, reqQoS);
  620 + processRpcSubscribe(grantedQoSList, topic, reqQoS, TopicType.V1);
  621 + activityReported = true;
  622 + break;
  623 + }
  624 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC: {
  625 + processRpcSubscribe(grantedQoSList, topic, reqQoS, TopicType.V2);
  626 + activityReported = true;
  627 + break;
  628 + }
  629 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC: {
  630 + processRpcSubscribe(grantedQoSList, topic, reqQoS, TopicType.V2_JSON);
  631 + activityReported = true;
  632 + break;
  633 + }
  634 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC: {
  635 + processRpcSubscribe(grantedQoSList, topic, reqQoS, TopicType.V2_PROTO);
552 636 activityReported = true;
553 637 break;
554 638 }
555 639 case MqttTopics.DEVICE_RPC_RESPONSE_SUB_TOPIC:
  640 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_TOPIC:
  641 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_JSON_TOPIC:
  642 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_PROTO_TOPIC:
556 643 case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC:
  644 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC:
  645 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_JSON_TOPIC:
  646 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC:
557 647 case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
558 648 case MqttTopics.GATEWAY_RPC_TOPIC:
559 649 case MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC:
... ... @@ -580,6 +670,18 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
580 670 ctx.writeAndFlush(createSubAckMessage(mqttMsg.variableHeader().messageId(), grantedQoSList));
581 671 }
582 672
  673 + private void processRpcSubscribe(List<Integer> grantedQoSList, String topic, MqttQoS reqQoS, TopicType topicType) {
  674 + transportService.process(deviceSessionCtx.getSessionInfo(), TransportProtos.SubscribeToRPCMsg.newBuilder().build(), null);
  675 + rpcSubTopicType = topicType;
  676 + registerSubQoS(topic, grantedQoSList, reqQoS);
  677 + }
  678 +
  679 + private void processAttributesSubscribe(List<Integer> grantedQoSList, String topic, MqttQoS reqQoS, TopicType topicType) {
  680 + transportService.process(deviceSessionCtx.getSessionInfo(), TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null);
  681 + attrSubTopicType = topicType;
  682 + registerSubQoS(topic, grantedQoSList, reqQoS);
  683 + }
  684 +
583 685 private void registerSubQoS(String topic, List<Integer> grantedQoSList, MqttQoS reqQoS) {
584 686 grantedQoSList.add(getMinSupportedQos(reqQoS));
585 687 mqttQoSMap.put(new MqttTopicMatcher(topic), getMinSupportedQos(reqQoS));
... ... @@ -595,18 +697,43 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
595 697 mqttQoSMap.remove(new MqttTopicMatcher(topicName));
596 698 try {
597 699 switch (topicName) {
598   - case MqttTopics.DEVICE_ATTRIBUTES_TOPIC: {
  700 + case MqttTopics.DEVICE_ATTRIBUTES_TOPIC:
  701 + case MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC:
  702 + case MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC:
  703 + case MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC: {
599 704 transportService.process(deviceSessionCtx.getSessionInfo(),
600 705 TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), null);
601 706 activityReported = true;
602 707 break;
603 708 }
604   - case MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC: {
  709 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC:
  710 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC:
  711 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC:
  712 + case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC: {
605 713 transportService.process(deviceSessionCtx.getSessionInfo(),
606 714 TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), null);
607 715 activityReported = true;
608 716 break;
609 717 }
  718 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_TOPIC:
  719 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_TOPIC:
  720 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_JSON_TOPIC:
  721 + case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_PROTO_TOPIC:
  722 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC:
  723 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC:
  724 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_JSON_TOPIC:
  725 + case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC:
  726 + case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
  727 + case MqttTopics.GATEWAY_RPC_TOPIC:
  728 + case MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC:
  729 + case MqttTopics.DEVICE_PROVISION_RESPONSE_TOPIC:
  730 + case MqttTopics.DEVICE_FIRMWARE_RESPONSES_TOPIC:
  731 + case MqttTopics.DEVICE_FIRMWARE_ERROR_TOPIC:
  732 + case MqttTopics.DEVICE_SOFTWARE_RESPONSES_TOPIC:
  733 + case MqttTopics.DEVICE_SOFTWARE_ERROR_TOPIC: {
  734 + activityReported = true;
  735 + break;
  736 + }
610 737 }
611 738 } catch (Exception e) {
612 739 log.warn("[{}] Failed to process unsubscription [{}] to [{}]", sessionId, mqttMsg.variableHeader().messageId(), topicName);
... ... @@ -837,8 +964,29 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
837 964
838 965 @Override
839 966 public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg response) {
  967 + log.trace("[{}] Received get attributes response", sessionId);
  968 + String topicBase;
  969 + MqttTransportAdaptor adaptor;
  970 + switch (attrReqTopicType) {
  971 + case V2:
  972 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  973 + topicBase = MqttTopics.DEVICE_ATTRIBUTES_RESPONSE_SHORT_TOPIC_PREFIX;
  974 + break;
  975 + case V2_JSON:
  976 + adaptor = context.getJsonMqttAdaptor();
  977 + topicBase = MqttTopics.DEVICE_ATTRIBUTES_RESPONSE_SHORT_JSON_TOPIC_PREFIX;
  978 + break;
  979 + case V2_PROTO:
  980 + adaptor = context.getProtoMqttAdaptor();
  981 + topicBase = MqttTopics.DEVICE_ATTRIBUTES_RESPONSE_SHORT_PROTO_TOPIC_PREFIX;
  982 + break;
  983 + default:
  984 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  985 + topicBase = MqttTopics.DEVICE_ATTRIBUTES_RESPONSE_TOPIC_PREFIX;
  986 + break;
  987 + }
840 988 try {
841   - deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, response).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
  989 + adaptor.convertToPublish(deviceSessionCtx, response, topicBase).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
842 990 } catch (Exception e) {
843 991 log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
844 992 }
... ... @@ -847,8 +995,29 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
847 995 @Override
848 996 public void onAttributeUpdate(UUID sessionId, TransportProtos.AttributeUpdateNotificationMsg notification) {
849 997 log.trace("[{}] Received attributes update notification to device", sessionId);
  998 + log.info("[{}] : attrSubTopicType: {}", notification.toString(), attrSubTopicType);
  999 + String topic;
  1000 + MqttTransportAdaptor adaptor;
  1001 + switch (attrSubTopicType) {
  1002 + case V2:
  1003 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  1004 + topic = MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC;
  1005 + break;
  1006 + case V2_JSON:
  1007 + adaptor = context.getJsonMqttAdaptor();
  1008 + topic = MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC;
  1009 + break;
  1010 + case V2_PROTO:
  1011 + adaptor = context.getProtoMqttAdaptor();
  1012 + topic = MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC;
  1013 + break;
  1014 + default:
  1015 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  1016 + topic = MqttTopics.DEVICE_ATTRIBUTES_TOPIC;
  1017 + break;
  1018 + }
850 1019 try {
851   - deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, notification).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
  1020 + adaptor.convertToPublish(deviceSessionCtx, notification, topic).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
852 1021 } catch (Exception e) {
853 1022 log.trace("[{}] Failed to convert device attributes update to MQTT msg", sessionId, e);
854 1023 }
... ... @@ -862,9 +1031,29 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
862 1031
863 1032 @Override
864 1033 public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) {
865   - log.trace("[{}] Received RPC command to device", sessionId);
  1034 + log.info("[{}] Received RPC command to device", sessionId);
  1035 + String baseTopic;
  1036 + MqttTransportAdaptor adaptor;
  1037 + switch (rpcSubTopicType) {
  1038 + case V2:
  1039 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  1040 + baseTopic = MqttTopics.DEVICE_RPC_REQUESTS_SHORT_TOPIC;
  1041 + break;
  1042 + case V2_JSON:
  1043 + adaptor = context.getJsonMqttAdaptor();
  1044 + baseTopic = MqttTopics.DEVICE_RPC_REQUESTS_SHORT_JSON_TOPIC;
  1045 + break;
  1046 + case V2_PROTO:
  1047 + adaptor = context.getProtoMqttAdaptor();
  1048 + baseTopic = MqttTopics.DEVICE_RPC_REQUESTS_SHORT_PROTO_TOPIC;
  1049 + break;
  1050 + default:
  1051 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  1052 + baseTopic = MqttTopics.DEVICE_RPC_REQUESTS_TOPIC;
  1053 + break;
  1054 + }
866 1055 try {
867   - deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> {
  1056 + adaptor.convertToPublish(deviceSessionCtx, rpcRequest, baseTopic).ifPresent(payload -> {
868 1057 int msgId = ((MqttPublishMessage) payload).variableHeader().packetId();
869 1058 if (isAckExpected(payload)) {
870 1059 rpcAwaitingAck.put(msgId, rpcRequest);
... ... @@ -898,9 +1087,29 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
898 1087
899 1088 @Override
900 1089 public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg rpcResponse) {
901   - log.trace("[{}] Received RPC command to server", sessionId);
  1090 + log.trace("[{}] Received RPC response from server", sessionId);
  1091 + String baseTopic;
  1092 + MqttTransportAdaptor adaptor;
  1093 + switch (toServerRpcSubTopicType) {
  1094 + case V2:
  1095 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  1096 + baseTopic = MqttTopics.DEVICE_RPC_RESPONSE_SHORT_TOPIC;
  1097 + break;
  1098 + case V2_JSON:
  1099 + adaptor = context.getJsonMqttAdaptor();
  1100 + baseTopic = MqttTopics.DEVICE_RPC_RESPONSE_SHORT_JSON_TOPIC;
  1101 + break;
  1102 + case V2_PROTO:
  1103 + adaptor = context.getProtoMqttAdaptor();
  1104 + baseTopic = MqttTopics.DEVICE_RPC_RESPONSE_SHORT_PROTO_TOPIC;
  1105 + break;
  1106 + default:
  1107 + adaptor = deviceSessionCtx.getPayloadAdaptor();
  1108 + baseTopic = MqttTopics.DEVICE_RPC_RESPONSE_TOPIC;
  1109 + break;
  1110 + }
902 1111 try {
903   - deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcResponse).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
  1112 + adaptor.convertToPublish(deviceSessionCtx, rpcResponse, baseTopic).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
904 1113 } catch (Exception e) {
905 1114 log.trace("[{}] Failed to convert device RPC command to MQTT msg", sessionId, e);
906 1115 }
... ... @@ -923,4 +1132,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
923 1132 public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
924 1133 deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt);
925 1134 }
  1135 +
  1136 + private enum TopicType {
  1137 + V1, V2, V2_JSON, V2_PROTO
  1138 + }
  1139 +
926 1140 }
... ...
... ... @@ -64,6 +64,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
64 64 try {
65 65 return JsonConverter.convertToTelemetryProto(new JsonParser().parse(payload));
66 66 } catch (IllegalStateException | JsonSyntaxException ex) {
  67 + log.warn("Failed to decode post telemetry request", ex);
67 68 throw new AdaptorException(ex);
68 69 }
69 70 }
... ... @@ -74,6 +75,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
74 75 try {
75 76 return JsonConverter.convertToAttributesProto(new JsonParser().parse(payload));
76 77 } catch (IllegalStateException | JsonSyntaxException ex) {
  78 + log.warn("Failed to decode post attributes request", ex);
77 79 throw new AdaptorException(ex);
78 80 }
79 81 }
... ... @@ -84,6 +86,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
84 86 try {
85 87 return JsonConverter.convertToClaimDeviceProto(ctx.getDeviceId(), payload);
86 88 } catch (IllegalStateException | JsonSyntaxException ex) {
  89 + log.warn("Failed to decode claim device request", ex);
87 90 throw new AdaptorException(ex);
88 91 }
89 92 }
... ... @@ -99,33 +102,33 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
99 102 }
100 103
101 104 @Override
102   - public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
103   - return processGetAttributeRequestMsg(inbound, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
  105 + public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topicBase) throws AdaptorException {
  106 + return processGetAttributeRequestMsg(inbound, topicBase);
104 107 }
105 108
106 109 @Override
107   - public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
108   - return processToDeviceRpcResponseMsg(inbound, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC);
  110 + public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topicBase) throws AdaptorException {
  111 + return processToDeviceRpcResponseMsg(inbound, topicBase);
109 112 }
110 113
111 114 @Override
112   - public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
113   - return processToServerRpcRequestMsg(ctx, inbound, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC);
  115 + public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topicBase) throws AdaptorException {
  116 + return processToServerRpcRequestMsg(ctx, inbound, topicBase);
114 117 }
115 118
116 119 @Override
117   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
118   - return processConvertFromAttributeResponseMsg(ctx, responseMsg, MqttTopics.DEVICE_ATTRIBUTES_RESPONSE_TOPIC_PREFIX);
  120 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg, String topicBase) throws AdaptorException {
  121 + return processConvertFromAttributeResponseMsg(ctx, responseMsg, topicBase);
119 122 }
120 123
121 124 @Override
122 125 public Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
123   - return processConvertFromGatewayAttributeResponseMsg(ctx, deviceName, responseMsg, MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC);
  126 + return processConvertFromGatewayAttributeResponseMsg(ctx, deviceName, responseMsg);
124 127 }
125 128
126 129 @Override
127   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) {
128   - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_ATTRIBUTES_TOPIC, JsonConverter.toJson(notificationMsg)));
  130 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg, String topic) {
  131 + return Optional.of(createMqttPublishMsg(ctx, topic, JsonConverter.toJson(notificationMsg)));
129 132 }
130 133
131 134 @Override
... ... @@ -135,8 +138,8 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
135 138 }
136 139
137 140 @Override
138   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) {
139   - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), JsonConverter.toJson(rpcRequest, false)));
  141 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, String topicBase) {
  142 + return Optional.of(createMqttPublishMsg(ctx, topicBase + rpcRequest.getRequestId(), JsonConverter.toJson(rpcRequest, false)));
140 143 }
141 144
142 145 @Override
... ... @@ -145,8 +148,8 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
145 148 }
146 149
147 150 @Override
148   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse) {
149   - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC + rpcResponse.getRequestId(), JsonConverter.toJson(rpcResponse)));
  151 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse, String topicBase) {
  152 + return Optional.of(createMqttPublishMsg(ctx, topicBase + rpcResponse.getRequestId(), JsonConverter.toJson(rpcResponse)));
150 153 }
151 154
152 155 @Override
... ... @@ -169,11 +172,11 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
169 172 }
170 173 }
171 174
172   - protected TransportProtos.GetAttributeRequestMsg processGetAttributeRequestMsg(MqttPublishMessage inbound, String topic) throws AdaptorException {
  175 + private TransportProtos.GetAttributeRequestMsg processGetAttributeRequestMsg(MqttPublishMessage inbound, String topicBase) throws AdaptorException {
173 176 String topicName = inbound.variableHeader().topicName();
174 177 try {
175 178 TransportProtos.GetAttributeRequestMsg.Builder result = TransportProtos.GetAttributeRequestMsg.newBuilder();
176   - result.setRequestId(getRequestId(topicName, topic));
  179 + result.setRequestId(getRequestId(topicName, topicBase));
177 180 String payload = inbound.payload().toString(UTF8);
178 181 JsonElement requestBody = new JsonParser().parse(payload);
179 182 Set<String> clientKeys = toStringSet(requestBody, "clientKeys");
... ... @@ -191,49 +194,50 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
191 194 }
192 195 }
193 196
194   - protected TransportProtos.ToDeviceRpcResponseMsg processToDeviceRpcResponseMsg(MqttPublishMessage inbound, String topic) throws AdaptorException {
  197 + private TransportProtos.ToDeviceRpcResponseMsg processToDeviceRpcResponseMsg(MqttPublishMessage inbound, String topicBase) throws AdaptorException {
195 198 String topicName = inbound.variableHeader().topicName();
196 199 try {
197   - int requestId = getRequestId(topicName, topic);
  200 + int requestId = getRequestId(topicName, topicBase);
198 201 String payload = inbound.payload().toString(UTF8);
199 202 return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId).setPayload(payload).build();
200 203 } catch (RuntimeException e) {
201   - log.warn("Failed to decode Rpc response", e);
  204 + log.warn("Failed to decode rpc response", e);
202 205 throw new AdaptorException(e);
203 206 }
204 207 }
205 208
206   - protected TransportProtos.ToServerRpcRequestMsg processToServerRpcRequestMsg(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topic) throws AdaptorException {
  209 + private TransportProtos.ToServerRpcRequestMsg processToServerRpcRequestMsg(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topicBase) throws AdaptorException {
207 210 String topicName = inbound.variableHeader().topicName();
208 211 String payload = validatePayload(ctx.getSessionId(), inbound.payload(), false);
209 212 try {
210   - int requestId = getRequestId(topicName, topic);
  213 + int requestId = getRequestId(topicName, topicBase);
211 214 return JsonConverter.convertToServerRpcRequest(new JsonParser().parse(payload), requestId);
212 215 } catch (IllegalStateException | JsonSyntaxException ex) {
  216 + log.warn("Failed to decode to server rpc request", ex);
213 217 throw new AdaptorException(ex);
214 218 }
215 219 }
216 220
217   - protected Optional<MqttMessage> processConvertFromAttributeResponseMsg(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg, String topic) throws AdaptorException {
  221 + private Optional<MqttMessage> processConvertFromAttributeResponseMsg(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg, String topicBase) throws AdaptorException {
218 222 if (!StringUtils.isEmpty(responseMsg.getError())) {
219 223 throw new AdaptorException(responseMsg.getError());
220 224 } else {
221 225 int requestId = responseMsg.getRequestId();
222 226 if (requestId >= 0) {
223 227 return Optional.of(createMqttPublishMsg(ctx,
224   - topic + requestId,
  228 + topicBase + requestId,
225 229 JsonConverter.toJson(responseMsg)));
226 230 }
227 231 return Optional.empty();
228 232 }
229 233 }
230 234
231   - protected Optional<MqttMessage> processConvertFromGatewayAttributeResponseMsg(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg, String topic) throws AdaptorException {
  235 + private Optional<MqttMessage> processConvertFromGatewayAttributeResponseMsg(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
232 236 if (!StringUtils.isEmpty(responseMsg.getError())) {
233 237 throw new AdaptorException(responseMsg.getError());
234 238 } else {
235 239 JsonObject result = JsonConverter.getJsonObjectForGateway(deviceName, responseMsg);
236   - return Optional.of(createMqttPublishMsg(ctx, topic, result));
  240 + return Optional.of(createMqttPublishMsg(ctx, MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, result));
237 241 }
238 242 }
239 243
... ...
... ... @@ -52,27 +52,27 @@ public interface MqttTransportAdaptor {
52 52
53 53 PostAttributeMsg convertToPostAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
54 54
55   - GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
  55 + GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topicBase) throws AdaptorException;
56 56
57   - ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
  57 + ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg, String topicBase) throws AdaptorException;
58 58
59   - ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
  59 + ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg, String topicBase) throws AdaptorException;
60 60
61 61 ClaimDeviceMsg convertToClaimDevice(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
62 62
63   - Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, GetAttributeResponseMsg responseMsg) throws AdaptorException;
  63 + Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, GetAttributeResponseMsg responseMsg, String topicBase) throws AdaptorException;
64 64
65 65 Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, GetAttributeResponseMsg responseMsg) throws AdaptorException;
66 66
67   - Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException;
  67 + Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, AttributeUpdateNotificationMsg notificationMsg, String topic) throws AdaptorException;
68 68
69 69 Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException;
70 70
71   - Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException;
  71 + Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, ToDeviceRpcRequestMsg rpcRequest, String topicBase) throws AdaptorException;
72 72
73 73 Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException;
74 74
75   - Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, ToServerRpcResponseMsg rpcResponse) throws AdaptorException;
  75 + Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, ToServerRpcResponseMsg rpcResponse, String topicBase) throws AdaptorException;
76 76
77 77 ProvisionDeviceRequestMsg convertToProvisionRequestMsg(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
78 78
... ...
... ... @@ -20,7 +20,6 @@ import com.google.gson.JsonParser;
20 20 import com.google.protobuf.Descriptors;
21 21 import com.google.protobuf.DynamicMessage;
22 22 import com.google.protobuf.InvalidProtocolBufferException;
23   -import com.google.protobuf.util.JsonFormat;
24 23 import io.netty.buffer.ByteBuf;
25 24 import io.netty.handler.codec.mqtt.MqttMessage;
26 25 import io.netty.handler.codec.mqtt.MqttPublishMessage;
... ... @@ -49,10 +48,11 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor {
49 48 public TransportProtos.PostTelemetryMsg convertToPostTelemetry(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
50 49 DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx;
51 50 byte[] bytes = toBytes(inbound.payload());
52   - Descriptors.Descriptor telemetryDynamicMsgDescriptor = getDescriptor(deviceSessionCtx.getTelemetryDynamicMsgDescriptor());
  51 + Descriptors.Descriptor telemetryDynamicMsgDescriptor = ProtoConverter.validateDescriptor(deviceSessionCtx.getTelemetryDynamicMsgDescriptor());
53 52 try {
54   - return JsonConverter.convertToTelemetryProto(new JsonParser().parse(dynamicMsgToJson(bytes, telemetryDynamicMsgDescriptor)));
  53 + return JsonConverter.convertToTelemetryProto(new JsonParser().parse(ProtoConverter.dynamicMsgToJson(bytes, telemetryDynamicMsgDescriptor)));
55 54 } catch (Exception e) {
  55 + log.warn("Failed to decode post telemetry request", e);
56 56 throw new AdaptorException(e);
57 57 }
58 58 }
... ... @@ -61,10 +61,11 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor {
61 61 public TransportProtos.PostAttributeMsg convertToPostAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
62 62 DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx;
63 63 byte[] bytes = toBytes(inbound.payload());
64   - Descriptors.Descriptor attributesDynamicMessageDescriptor = getDescriptor(deviceSessionCtx.getAttributesDynamicMessageDescriptor());
  64 + Descriptors.Descriptor attributesDynamicMessageDescriptor = ProtoConverter.validateDescriptor(deviceSessionCtx.getAttributesDynamicMessageDescriptor());
65 65 try {
66   - return JsonConverter.convertToAttributesProto(new JsonParser().parse(dynamicMsgToJson(bytes, attributesDynamicMessageDescriptor)));
  66 + return JsonConverter.convertToAttributesProto(new JsonParser().parse(ProtoConverter.dynamicMsgToJson(bytes, attributesDynamicMessageDescriptor)));
67 67 } catch (Exception e) {
  68 + log.warn("Failed to decode post attributes request", e);
68 69 throw new AdaptorException(e);
69 70 }
70 71 }
... ... @@ -75,16 +76,17 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor {
75 76 try {
76 77 return ProtoConverter.convertToClaimDeviceProto(ctx.getDeviceId(), bytes);
77 78 } catch (InvalidProtocolBufferException e) {
  79 + log.warn("Failed to decode claim device request", e);
78 80 throw new AdaptorException(e);
79 81 }
80 82 }
81 83
82 84 @Override
83   - public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
  85 + public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound, String topicBase) throws AdaptorException {
84 86 byte[] bytes = toBytes(inbound.payload());
85 87 String topicName = inbound.variableHeader().topicName();
86 88 try {
87   - int requestId = getRequestId(topicName, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX);
  89 + int requestId = getRequestId(topicName, topicBase);
88 90 return ProtoConverter.convertToGetAttributeRequestMessage(bytes, requestId);
89 91 } catch (InvalidProtocolBufferException e) {
90 92 log.warn("Failed to decode get attributes request", e);
... ... @@ -93,29 +95,30 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor {
93 95 }
94 96
95 97 @Override
96   - public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException {
  98 + public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg, String topicBase) throws AdaptorException {
97 99 DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx;
98 100 String topicName = mqttMsg.variableHeader().topicName();
99 101 byte[] bytes = toBytes(mqttMsg.payload());
100   - Descriptors.Descriptor rpcResponseDynamicMessageDescriptor = getDescriptor(deviceSessionCtx.getRpcResponseDynamicMessageDescriptor());
  102 + Descriptors.Descriptor rpcResponseDynamicMessageDescriptor = ProtoConverter.validateDescriptor(deviceSessionCtx.getRpcResponseDynamicMessageDescriptor());
101 103 try {
102   - int requestId = getRequestId(topicName, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC);
103   - JsonElement response = new JsonParser().parse(dynamicMsgToJson(bytes, rpcResponseDynamicMessageDescriptor));
  104 + int requestId = getRequestId(topicName, topicBase);
  105 + JsonElement response = new JsonParser().parse(ProtoConverter.dynamicMsgToJson(bytes, rpcResponseDynamicMessageDescriptor));
104 106 return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId).setPayload(response.toString()).build();
105 107 } catch (Exception e) {
106   - log.warn("Failed to decode Rpc response", e);
  108 + log.warn("Failed to decode rpc response", e);
107 109 throw new AdaptorException(e);
108 110 }
109 111 }
110 112
111 113 @Override
112   - public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException {
  114 + public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg, String topicBase) throws AdaptorException {
113 115 byte[] bytes = toBytes(mqttMsg.payload());
114 116 String topicName = mqttMsg.variableHeader().topicName();
115 117 try {
116   - int requestId = getRequestId(topicName, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC);
  118 + int requestId = getRequestId(topicName, topicBase);
117 119 return ProtoConverter.convertToServerRpcRequest(bytes, requestId);
118 120 } catch (InvalidProtocolBufferException e) {
  121 + log.warn("Failed to decode to server rpc request", e);
119 122 throw new AdaptorException(e);
120 123 }
121 124 }
... ... @@ -126,40 +129,43 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor {
126 129 try {
127 130 return ProtoConverter.convertToProvisionRequestMsg(bytes);
128 131 } catch (InvalidProtocolBufferException ex) {
  132 + log.warn("Failed to decode provision request", ex);
129 133 throw new AdaptorException(ex);
130 134 }
131 135 }
132 136
133 137 @Override
134   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
  138 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg, String topicBase) throws AdaptorException {
135 139 if (!StringUtils.isEmpty(responseMsg.getError())) {
136 140 throw new AdaptorException(responseMsg.getError());
137 141 } else {
138 142 int requestId = responseMsg.getRequestId();
139 143 if (requestId >= 0) {
140   - return Optional.of(createMqttPublishMsg(ctx,
141   - MqttTopics.DEVICE_ATTRIBUTES_RESPONSE_TOPIC_PREFIX + requestId,
142   - responseMsg.toByteArray()));
  144 + return Optional.of(createMqttPublishMsg(ctx, topicBase + requestId, responseMsg.toByteArray()));
143 145 }
144 146 return Optional.empty();
145 147 }
146 148 }
147 149
148 150 @Override
149   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException {
  151 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, String topicBase) throws AdaptorException {
150 152 DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx;
151 153 DynamicMessage.Builder rpcRequestDynamicMessageBuilder = deviceSessionCtx.getRpcRequestDynamicMessageBuilder();
152   - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), ProtoConverter.convertToRpcRequest(rpcRequest, rpcRequestDynamicMessageBuilder)));
  154 + if (rpcRequestDynamicMessageBuilder == null) {
  155 + throw new AdaptorException("Failed to get rpcRequestDynamicMessageBuilder!");
  156 + } else {
  157 + return Optional.of(createMqttPublishMsg(ctx, topicBase + rpcRequest.getRequestId(), ProtoConverter.convertToRpcRequest(rpcRequest, rpcRequestDynamicMessageBuilder)));
  158 + }
153 159 }
154 160
155 161 @Override
156   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse) {
157   - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC + rpcResponse.getRequestId(), rpcResponse.toByteArray()));
  162 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse, String topicBase) {
  163 + return Optional.of(createMqttPublishMsg(ctx, topicBase + rpcResponse.getRequestId(), rpcResponse.toByteArray()));
158 164 }
159 165
160 166 @Override
161   - public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) {
162   - return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_ATTRIBUTES_TOPIC, notificationMsg.toByteArray()));
  167 + public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg, String topic) {
  168 + return Optional.of(createMqttPublishMsg(ctx, topic, notificationMsg.toByteArray()));
163 169 }
164 170
165 171 @Override
... ... @@ -213,17 +219,4 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor {
213 219 private int getRequestId(String topicName, String topic) {
214 220 return Integer.parseInt(topicName.substring(topic.length()));
215 221 }
216   -
217   - private Descriptors.Descriptor getDescriptor(Descriptors.Descriptor descriptor) throws AdaptorException {
218   - if (descriptor == null) {
219   - throw new AdaptorException("Failed to get dynamic message descriptor!");
220   - }
221   - return descriptor;
222   - }
223   -
224   - private String dynamicMsgToJson(byte[] bytes, Descriptors.Descriptor descriptor) throws InvalidProtocolBufferException {
225   - DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor, bytes);
226   - return JsonFormat.printer().includingDefaultValueFields().print(dynamicMessage);
227   - }
228   -
229 222 }
... ...
... ... @@ -19,6 +19,7 @@ import com.google.gson.Gson;
19 19 import com.google.gson.JsonElement;
20 20 import com.google.gson.JsonObject;
21 21 import com.google.gson.JsonParser;
  22 +import com.google.protobuf.Descriptors;
22 23 import com.google.protobuf.DynamicMessage;
23 24 import com.google.protobuf.InvalidProtocolBufferException;
24 25 import com.google.protobuf.util.JsonFormat;
... ... @@ -190,4 +191,17 @@ public class ProtoConverter {
190 191 throw new AdaptorException("Failed to convert ToDeviceRpcRequestMsg to Dynamic Rpc request message due to: ", e);
191 192 }
192 193 }
  194 +
  195 + public static Descriptors.Descriptor validateDescriptor(Descriptors.Descriptor descriptor) throws AdaptorException {
  196 + if (descriptor == null) {
  197 + throw new AdaptorException("Failed to get dynamic message descriptor!");
  198 + }
  199 + return descriptor;
  200 + }
  201 +
  202 + public static String dynamicMsgToJson(byte[] bytes, Descriptors.Descriptor descriptor) throws InvalidProtocolBufferException {
  203 + DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor, bytes);
  204 + return JsonFormat.printer().includingDefaultValueFields().print(dynamicMessage);
  205 + }
  206 +
193 207 }
... ...