ViacheslavKlimov
4 years ago
324 changed files with 12723 additions and 888 deletions
@ -0,0 +1,88 @@ |
|||
# |
|||
# Copyright © 2016-2022 The Thingsboard Authors |
|||
# |
|||
# Licensed under the Apache License, Version 2.0 (the "License"); |
|||
# you may not use this file except in compliance with the License. |
|||
# You may obtain a copy of the License at |
|||
# |
|||
# http://www.apache.org/licenses/LICENSE-2.0 |
|||
# |
|||
# Unless required by applicable law or agreed to in writing, software |
|||
# distributed under the License is distributed on an "AS IS" BASIS, |
|||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
# See the License for the specific language governing permissions and |
|||
# limitations under the License. |
|||
# |
|||
|
|||
changelog: |
|||
exclude: |
|||
labels: |
|||
- Ignore for release |
|||
categories: |
|||
- title: 'Major Core & Rule Engine' |
|||
labels: |
|||
- 'Major Core' |
|||
- 'Major Rule Engine' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Major UI' |
|||
labels: |
|||
- 'Major UI' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Major Transport' |
|||
labels: |
|||
- 'Major Transport' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Major Edge' |
|||
labels: |
|||
- 'Major Edge' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Core & Rule Engine' |
|||
labels: |
|||
- 'Core' |
|||
- 'Rule Engine' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'UI' |
|||
labels: |
|||
- 'UI' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Transport' |
|||
labels: |
|||
- 'Transport' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Edge' |
|||
labels: |
|||
- 'Edge' |
|||
exclude: |
|||
labels: |
|||
- 'Bug' |
|||
- title: 'Bug: Core & Rule Engine' |
|||
labels: |
|||
- 'Core' |
|||
- 'Rule Engine' |
|||
- 'Bug' |
|||
- title: 'Bug: UI' |
|||
labels: |
|||
- 'UI' |
|||
- 'Bug' |
|||
- title: 'Bug: Transport' |
|||
labels: |
|||
- 'Transport' |
|||
- 'Bug' |
|||
- title: 'Bug: Edge' |
|||
labels: |
|||
- 'Edge' |
|||
- 'Bug' |
|||
4
application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/MqttAttributesRequestBackwardCompatibilityIntegrationTest.java → application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestBackwardCompatibilityIntegrationTest.java
4
application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/request/MqttAttributesRequestBackwardCompatibilityIntegrationTest.java → application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestBackwardCompatibilityIntegrationTest.java
4
application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/updates/MqttAttributesUpdatesBackwardCompatibilityIntegrationTest.java → application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/updates/MqttAttributesUpdatesBackwardCompatibilityIntegrationTest.java
4
application/src/test/java/org/thingsboard/server/transport/mqtt/attributes/updates/MqttAttributesUpdatesBackwardCompatibilityIntegrationTest.java → application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/updates/MqttAttributesUpdatesBackwardCompatibilityIntegrationTest.java
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv3.client; |
|||
|
|||
import org.eclipse.paho.client.mqttv3.MqttException; |
|||
import org.junit.Assert; |
|||
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; |
|||
import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; |
|||
|
|||
public abstract class AbstractMqttClientConnectionTest extends AbstractMqttIntegrationTest { |
|||
|
|||
protected void processClientWithCorrectAccessTokenTest() throws Exception { |
|||
MqttTestClient client = new MqttTestClient(); |
|||
client.connectAndWait(accessToken); |
|||
Assert.assertTrue(client.isConnected()); |
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processClientWithWrongAccessTokenTest() throws Exception { |
|||
MqttTestClient client = new MqttTestClient(); |
|||
try { |
|||
client.connectAndWait("wrongAccessToken"); |
|||
} catch (MqttException e) { |
|||
Assert.assertEquals(MqttException.REASON_CODE_FAILED_AUTHENTICATION, e.getReasonCode()); |
|||
} |
|||
} |
|||
|
|||
protected void processClientWithWrongClientIdAndEmptyUsernamePasswordTest() throws Exception { |
|||
MqttTestClient client = new MqttTestClient("unknownClientId"); |
|||
try { |
|||
client.connectAndWait(); |
|||
} catch (MqttException e) { |
|||
Assert.assertEquals(MqttException.REASON_CODE_INVALID_CLIENT_ID, e.getReasonCode()); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv3.client; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test MqttV5 client device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithCorrectAccessToken() throws Exception { |
|||
processClientWithCorrectAccessTokenTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithWrongAccessToken() throws Exception { |
|||
processClientWithWrongAccessTokenTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithWrongClientIdAndEmptyUsernamePassword() throws Exception { |
|||
processClientWithWrongClientIdAndEmptyUsernamePasswordTest(); |
|||
} |
|||
} |
|||
@ -0,0 +1,21 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5; |
|||
|
|||
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; |
|||
|
|||
public abstract class AbstractMqttV5Test extends AbstractMqttIntegrationTest { |
|||
} |
|||
@ -0,0 +1,110 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5; |
|||
|
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.paho.mqttv5.client.IMqttToken; |
|||
import org.eclipse.paho.mqttv5.client.MqttCallback; |
|||
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; |
|||
import org.eclipse.paho.mqttv5.common.MqttException; |
|||
import org.eclipse.paho.mqttv5.common.MqttMessage; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttProperties; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage; |
|||
|
|||
import java.util.concurrent.CountDownLatch; |
|||
|
|||
@Slf4j |
|||
@Data |
|||
public class MqttV5TestCallback implements MqttCallback { |
|||
|
|||
protected CountDownLatch subscribeLatch; |
|||
protected final CountDownLatch deliveryLatch; |
|||
protected int qoS; |
|||
protected byte[] payloadBytes; |
|||
protected String awaitSubTopic; |
|||
protected boolean pubAckReceived; |
|||
protected MqttMessage lastReceivedMessage; |
|||
|
|||
public MqttV5TestCallback() { |
|||
this.subscribeLatch = new CountDownLatch(1); |
|||
this.deliveryLatch = new CountDownLatch(1); |
|||
} |
|||
|
|||
public MqttV5TestCallback(int subscribeCount) { |
|||
this.subscribeLatch = new CountDownLatch(subscribeCount); |
|||
this.deliveryLatch = new CountDownLatch(1); |
|||
} |
|||
|
|||
public MqttV5TestCallback(String awaitSubTopic) { |
|||
this.subscribeLatch = new CountDownLatch(1); |
|||
this.deliveryLatch = new CountDownLatch(1); |
|||
this.awaitSubTopic = awaitSubTopic; |
|||
} |
|||
|
|||
@Override |
|||
public void disconnected(MqttDisconnectResponse mqttDisconnectResponse) { |
|||
if (mqttDisconnectResponse.getException() != null) { |
|||
log.warn("connectionLost: ", mqttDisconnectResponse.getException()); |
|||
deliveryLatch.countDown(); |
|||
} |
|||
log.warn("Disconnected with reason: {}", mqttDisconnectResponse.getReasonString()); |
|||
} |
|||
|
|||
@Override |
|||
public void mqttErrorOccurred(MqttException e) { |
|||
log.warn("Error occurred:", e); |
|||
} |
|||
|
|||
@Override |
|||
public void messageArrived(String requestTopic, MqttMessage mqttMessage) { |
|||
lastReceivedMessage = mqttMessage; |
|||
if (awaitSubTopic == null) { |
|||
log.warn("messageArrived on topic: {}", requestTopic); |
|||
qoS = mqttMessage.getQos(); |
|||
payloadBytes = mqttMessage.getPayload(); |
|||
subscribeLatch.countDown(); |
|||
} else { |
|||
messageArrivedOnAwaitSubTopic(requestTopic, mqttMessage); |
|||
} |
|||
} |
|||
|
|||
protected void messageArrivedOnAwaitSubTopic(String requestTopic, MqttMessage mqttMessage) { |
|||
log.warn("messageArrived on topic: {}, awaitSubTopic: {}", requestTopic, awaitSubTopic); |
|||
if (awaitSubTopic.equals(requestTopic)) { |
|||
qoS = mqttMessage.getQos(); |
|||
payloadBytes = mqttMessage.getPayload(); |
|||
subscribeLatch.countDown(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void deliveryComplete(IMqttToken iMqttToken) { |
|||
log.warn("delivery complete: {}", iMqttToken.getResponse()); |
|||
pubAckReceived = iMqttToken.getResponse().getType() == MqttWireMessage.MESSAGE_TYPE_PUBACK; |
|||
deliveryLatch.countDown(); |
|||
} |
|||
|
|||
@Override |
|||
public void connectComplete(boolean reconnect, String serverURI) { |
|||
log.warn("Connect completed: reconnect - {}, serverURI - {}", reconnect, serverURI); |
|||
} |
|||
|
|||
@Override |
|||
public void authPacketArrived(int reasonCode, MqttProperties mqttProperties) { |
|||
log.warn("Auth package received: reasonCode - {}, mqtt properties - {}", reasonCode, mqttProperties); |
|||
} |
|||
} |
|||
@ -0,0 +1,175 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import org.eclipse.paho.mqttv5.client.IMqttToken; |
|||
import org.eclipse.paho.mqttv5.client.MqttAsyncClient; |
|||
import org.eclipse.paho.mqttv5.client.MqttCallback; |
|||
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; |
|||
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence; |
|||
import org.eclipse.paho.mqttv5.common.MqttException; |
|||
import org.eclipse.paho.mqttv5.common.MqttMessage; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
public class MqttV5TestClient { // We should copy part of MqttV3TestClient, due to different package names in import
|
|||
|
|||
private static final String MQTT_URL = "tcp://localhost:1883"; |
|||
private static final int TIMEOUT = 30; // seconds
|
|||
private static final long TIMEOUT_MS = TimeUnit.SECONDS.toMillis(TIMEOUT); |
|||
|
|||
private final MqttAsyncClient client; |
|||
|
|||
public void setCallback(MqttCallback callback) { |
|||
client.setCallback(callback); |
|||
} |
|||
|
|||
public MqttV5TestClient() throws MqttException { |
|||
this.client = createClient(); |
|||
} |
|||
|
|||
public MqttV5TestClient(String clientId) throws MqttException { |
|||
this.client = createClient(clientId); |
|||
} |
|||
|
|||
public MqttV5TestClient(boolean generateClientId) throws MqttException { |
|||
this.client = createClient(generateClientId); |
|||
} |
|||
|
|||
public IMqttToken connectAndWait(String userName, String password) throws MqttException { |
|||
IMqttToken connect = connect(userName, password); |
|||
connect.waitForCompletion(TIMEOUT_MS); |
|||
return connect; |
|||
} |
|||
|
|||
public IMqttToken connectAndWait(String userName) throws MqttException { |
|||
return connectAndWait(userName, null); |
|||
} |
|||
|
|||
public IMqttToken connectAndWait() throws MqttException { |
|||
return connectAndWait(null, null); |
|||
} |
|||
|
|||
public IMqttToken connectAndWait(MqttConnectionOptions options) throws MqttException { |
|||
IMqttToken iMqttToken = connect(options); |
|||
iMqttToken.waitForCompletion(TIMEOUT_MS); |
|||
return iMqttToken; |
|||
} |
|||
|
|||
private IMqttToken connect(String userName, String password) throws MqttException { |
|||
if (client == null) { |
|||
throw new RuntimeException("Failed to connect! MqttAsyncClient is not initialized!"); |
|||
} |
|||
MqttConnectionOptions options = new MqttConnectionOptions(); |
|||
if (StringUtils.isNotEmpty(userName)) { |
|||
options.setUserName(userName); |
|||
} |
|||
if (StringUtils.isNotEmpty(password)) { |
|||
options.setPassword(password.getBytes()); |
|||
} |
|||
return client.connect(options); |
|||
} |
|||
|
|||
public IMqttToken connect(MqttConnectionOptions options) throws MqttException { |
|||
if (client == null) { |
|||
throw new RuntimeException("Failed to connect! MqttAsyncClient is not initialized!"); |
|||
} |
|||
return client.connect(options); |
|||
} |
|||
|
|||
public void disconnectAndWait() throws MqttException { |
|||
disconnect().waitForCompletion(TIMEOUT_MS); |
|||
} |
|||
|
|||
public IMqttToken disconnect() throws MqttException { |
|||
return client.disconnect(); |
|||
} |
|||
|
|||
public void disconnectForcibly() throws MqttException { |
|||
client.disconnectForcibly(TIMEOUT_MS); |
|||
} |
|||
|
|||
public IMqttToken publishAndWait(String topic, byte[] payload) throws MqttException { |
|||
IMqttToken iMqttToken = publish(topic, payload); |
|||
iMqttToken.waitForCompletion(TIMEOUT_MS); |
|||
return iMqttToken; |
|||
} |
|||
|
|||
public IMqttToken publish(String topic, byte[] payload) throws MqttException { |
|||
MqttMessage message = new MqttMessage(); |
|||
message.setPayload(payload); |
|||
return publish(topic, message); |
|||
} |
|||
|
|||
public IMqttToken publish(String topic, MqttMessage message) throws MqttException { |
|||
return publish(topic, message.getPayload(), message.getQos(), message.isRetained()); |
|||
} |
|||
|
|||
public IMqttToken publish(String topic, byte[] payload, int qos, boolean retain) throws MqttException { |
|||
return client.publish(topic, payload, qos, retain); |
|||
} |
|||
|
|||
public IMqttToken subscribeAndWait(String topic, MqttQoS qoS) throws MqttException { |
|||
IMqttToken iMqttToken = subscribe(topic, qoS); |
|||
iMqttToken.waitForCompletion(TIMEOUT_MS); |
|||
return iMqttToken; |
|||
} |
|||
|
|||
public IMqttToken subscribe(String topic, MqttQoS qoS) throws MqttException { |
|||
return client.subscribe(topic, qoS.value()); |
|||
} |
|||
|
|||
public IMqttToken unsubscribeAndWait(String topic) throws MqttException { |
|||
IMqttToken iMqttToken = unsubscribe(topic); |
|||
iMqttToken.waitForCompletion(TIMEOUT_MS); |
|||
return iMqttToken; |
|||
} |
|||
|
|||
public IMqttToken unsubscribe(String topic) throws MqttException { |
|||
return client.unsubscribe(topic); |
|||
} |
|||
|
|||
public boolean isConnected() { |
|||
return client.isConnected(); |
|||
} |
|||
|
|||
public void enableManualAcks() { |
|||
client.setManualAcks(true); |
|||
} |
|||
|
|||
public void messageArrivedComplete(MqttMessage mqttMessage) throws MqttException { |
|||
client.messageArrivedComplete(mqttMessage.getId(), mqttMessage.getQos()); |
|||
} |
|||
|
|||
private MqttAsyncClient createClient() throws MqttException { |
|||
return createClient(true); |
|||
} |
|||
|
|||
private MqttAsyncClient createClient(boolean generateClientId) throws MqttException { |
|||
String clientId = null; |
|||
if (generateClientId) { |
|||
clientId = "test" + System.nanoTime(); |
|||
} |
|||
return createClient(clientId); |
|||
} |
|||
|
|||
private MqttAsyncClient createClient(String clientId) throws MqttException { |
|||
return new MqttAsyncClient(MQTT_URL, clientId, new MemoryPersistence()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,161 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.attributes; |
|||
|
|||
import com.fasterxml.jackson.core.type.TypeReference; |
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import org.junit.Before; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.device.profile.MqttTopics; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import java.util.Arrays; |
|||
import java.util.HashSet; |
|||
import java.util.LinkedHashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.junit.Assert.assertEquals; |
|||
import static org.junit.Assert.assertNotNull; |
|||
import static org.junit.Assert.assertTrue; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
|
|||
public abstract class AbstractAttributesMqttV5Test extends AbstractMqttV5Test { |
|||
|
|||
private static final String SHARED_ATTRIBUTES_PAYLOAD = "{\"sharedStr\":\"value1\",\"sharedBool\":true,\"sharedDbl\":42.0,\"sharedLong\":73," + |
|||
"\"sharedJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}"; |
|||
|
|||
private static final String SHARED_ATTRIBUTES_DELETED_RESPONSE = "{\"deleted\":[\"sharedJson\"]}"; |
|||
|
|||
protected static final String PAYLOAD_VALUES_STR = "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4," + |
|||
" \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test Post Attributes device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
protected void processAttributesPublishTest() throws Exception { |
|||
List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); |
|||
|
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
client.publishAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, PAYLOAD_VALUES_STR.getBytes()); |
|||
client.disconnectAndWait(); |
|||
|
|||
DeviceId deviceId = savedDevice.getId(); |
|||
|
|||
long start = System.currentTimeMillis(); |
|||
long end = System.currentTimeMillis() + 5000; |
|||
|
|||
List<String> actualKeys = null; |
|||
while (start <= end) { |
|||
actualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/attributes/CLIENT_SCOPE", new TypeReference<>() { |
|||
}); |
|||
if (actualKeys.size() == expectedKeys.size()) { |
|||
break; |
|||
} |
|||
Thread.sleep(100); |
|||
start += 100; |
|||
} |
|||
assertNotNull(actualKeys); |
|||
|
|||
Set<String> actualKeySet = new HashSet<>(actualKeys); |
|||
|
|||
Set<String> expectedKeySet = new HashSet<>(expectedKeys); |
|||
|
|||
assertEquals(expectedKeySet, actualKeySet); |
|||
|
|||
String getAttributesValuesUrl = getAttributesValuesUrl(deviceId, actualKeySet); |
|||
List<Map<String, Object>> values = doGetAsyncTyped(getAttributesValuesUrl, new TypeReference<>() { |
|||
}); |
|||
assertAttributesValues(values, actualKeySet); |
|||
String deleteAttributesUrl = "/api/plugins/telemetry/DEVICE/" + deviceId + "/CLIENT_SCOPE?keys=" + String.join(",", actualKeySet); |
|||
doDelete(deleteAttributesUrl); |
|||
} |
|||
|
|||
protected void processAttributesUpdatesTest() throws Exception { |
|||
|
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
MqttV5TestCallback onUpdateCallback = new MqttV5TestCallback(); |
|||
client.setCallback(onUpdateCallback); |
|||
client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE); |
|||
|
|||
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
|||
onUpdateCallback.getSubscribeLatch().await(3, TimeUnit.SECONDS); |
|||
|
|||
validateUpdateAttributesResponse(onUpdateCallback, SHARED_ATTRIBUTES_PAYLOAD); |
|||
|
|||
MqttV5TestCallback onDeleteCallback = new MqttV5TestCallback(); |
|||
client.setCallback(onDeleteCallback); |
|||
doDelete("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/SHARED_SCOPE?keys=sharedJson", String.class); |
|||
onDeleteCallback.getSubscribeLatch().await(3, TimeUnit.SECONDS); |
|||
|
|||
validateUpdateAttributesResponse(onDeleteCallback, SHARED_ATTRIBUTES_DELETED_RESPONSE); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
private String getAttributesValuesUrl(DeviceId deviceId, Set<String> actualKeySet) { |
|||
return "/api/plugins/telemetry/DEVICE/" + deviceId + "/values/attributes/CLIENT_SCOPE?keys=" + String.join(",", actualKeySet); |
|||
} |
|||
|
|||
protected void assertAttributesValues(List<Map<String, Object>> deviceValues, Set<String> keySet) { |
|||
for (Map<String, Object> map : deviceValues) { |
|||
String key = (String) map.get("key"); |
|||
Object value = map.get("value"); |
|||
assertTrue(keySet.contains(key)); |
|||
switch (key) { |
|||
case "key1": |
|||
assertEquals("value1", value); |
|||
break; |
|||
case "key2": |
|||
assertEquals(true, value); |
|||
break; |
|||
case "key3": |
|||
assertEquals(3.0, value); |
|||
break; |
|||
case "key4": |
|||
assertEquals(4, value); |
|||
break; |
|||
case "key5": |
|||
assertNotNull(value); |
|||
assertEquals(3, ((LinkedHashMap) value).size()); |
|||
assertEquals(42, ((LinkedHashMap) value).get("someNumber")); |
|||
assertEquals(Arrays.asList(1, 2, 3), ((LinkedHashMap) value).get("someArray")); |
|||
LinkedHashMap<String, String> someNestedObject = (LinkedHashMap) ((LinkedHashMap) value).get("someNestedObject"); |
|||
assertEquals("value", someNestedObject.get("key")); |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
protected void validateUpdateAttributesResponse(MqttV5TestCallback callback, String expectedResponse) { |
|||
assertNotNull(callback.getPayloadBytes()); |
|||
assertEquals(JacksonUtil.toJsonNode(expectedResponse), JacksonUtil.fromBytes(callback.getPayloadBytes())); |
|||
} |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.attributes.updates; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.common.data.TransportPayloadType; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.attributes.AbstractAttributesMqttV5Test; |
|||
|
|||
@Slf4j |
|||
@DaoSqlTest |
|||
public class AttributesUpdatesTest extends AbstractAttributesMqttV5Test { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test Subscribe to attribute updates") |
|||
.transportPayloadType(TransportPayloadType.JSON) |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testAttributeMqttV5SimpleClientUpdates() throws Exception { |
|||
processAttributesUpdatesTest(); |
|||
} |
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.attributes.upload; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.attributes.AbstractAttributesMqttV5Test; |
|||
|
|||
@DaoSqlTest |
|||
public class AttributesPublishTest extends AbstractAttributesMqttV5Test { |
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test Post Attributes device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testMqttV5AttributePublishTest() throws Exception { |
|||
processAttributesPublishTest(); |
|||
} |
|||
} |
|||
@ -0,0 +1,101 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.claim; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.data.ClaimRequest; |
|||
import org.thingsboard.server.common.data.Customer; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.security.Authority; |
|||
import org.thingsboard.server.dao.device.claim.ClaimResponse; |
|||
import org.thingsboard.server.dao.device.claim.ClaimResult; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import static org.junit.Assert.assertEquals; |
|||
import static org.junit.Assert.assertNotNull; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_CLAIM_TOPIC; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractMqttV5ClaimTest extends AbstractMqttV5Test { |
|||
protected static final String CUSTOMER_USER_PASSWORD = "customerUser123!"; |
|||
|
|||
protected User customerAdmin; |
|||
protected Customer savedCustomer; |
|||
|
|||
protected void processTestClaimingDevice() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
byte[] payloadBytes; |
|||
byte[] failurePayloadBytes; |
|||
payloadBytes = "{\"secretKey\":\"value\", \"durationMs\":60000}".getBytes(); |
|||
failurePayloadBytes = "{\"secretKey\":\"value\", \"durationMs\":1}".getBytes(); |
|||
validateClaimResponse(client, payloadBytes, failurePayloadBytes); |
|||
} |
|||
|
|||
protected void validateClaimResponse(MqttV5TestClient client, byte[] payloadBytes, byte[] failurePayloadBytes) throws Exception { |
|||
client.publishAndWait(DEVICE_CLAIM_TOPIC, failurePayloadBytes); |
|||
|
|||
loginUser(customerAdmin.getName(), CUSTOMER_USER_PASSWORD); |
|||
ClaimRequest claimRequest = new ClaimRequest("value"); |
|||
|
|||
ClaimResponse claimResponse = doExecuteWithRetriesAndInterval( |
|||
() -> doPostClaimAsync("/api/customer/device/" + savedDevice.getName() + "/claim", claimRequest, ClaimResponse.class, status().isBadRequest()), |
|||
20, |
|||
100 |
|||
); |
|||
|
|||
assertEquals(claimResponse, ClaimResponse.FAILURE); |
|||
|
|||
client.publishAndWait(DEVICE_CLAIM_TOPIC, payloadBytes); |
|||
client.disconnect(); |
|||
|
|||
ClaimResult claimResult = doExecuteWithRetriesAndInterval( |
|||
() -> doPostClaimAsync("/api/customer/device/" + savedDevice.getName() + "/claim", claimRequest, ClaimResult.class, status().isOk()), |
|||
20, |
|||
100 |
|||
); |
|||
assertEquals(claimResult.getResponse(), ClaimResponse.SUCCESS); |
|||
Device claimedDevice = claimResult.getDevice(); |
|||
assertNotNull(claimedDevice); |
|||
assertNotNull(claimedDevice.getCustomerId()); |
|||
assertEquals(customerAdmin.getCustomerId(), claimedDevice.getCustomerId()); |
|||
|
|||
claimResponse = doPostClaimAsync("/api/customer/device/" + savedDevice.getName() + "/claim", claimRequest, ClaimResponse.class, status().isBadRequest()); |
|||
assertEquals(claimResponse, ClaimResponse.CLAIMED); |
|||
} |
|||
|
|||
protected void createCustomerAndUser() throws Exception { |
|||
Customer customer = new Customer(); |
|||
customer.setTenantId(tenantId); |
|||
customer.setTitle("Test Claiming Customer"); |
|||
savedCustomer = doPost("/api/customer", customer, Customer.class); |
|||
assertNotNull(savedCustomer); |
|||
assertEquals(tenantId, savedCustomer.getTenantId()); |
|||
|
|||
User user = new User(); |
|||
user.setAuthority(Authority.CUSTOMER_USER); |
|||
user.setTenantId(tenantId); |
|||
user.setCustomerId(savedCustomer.getId()); |
|||
user.setEmail("customer@thingsboard.org"); |
|||
|
|||
customerAdmin = createUser(user, CUSTOMER_USER_PASSWORD); |
|||
assertNotNull(customerAdmin); |
|||
assertEquals(customerAdmin.getCustomerId(), savedCustomer.getId()); |
|||
} |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.claim; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5ClaimTest extends AbstractMqttV5ClaimTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test Claim device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
createCustomerAndUser(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClaimingDevice() throws Exception { |
|||
processTestClaimingDevice(); |
|||
} |
|||
} |
|||
@ -0,0 +1,111 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.connection; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.eclipse.paho.mqttv5.client.IMqttToken; |
|||
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; |
|||
import org.eclipse.paho.mqttv5.common.MqttException; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttConnAck; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage; |
|||
import org.junit.Assert; |
|||
import org.thingsboard.server.common.data.device.profile.MqttTopics; |
|||
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_CONNACK; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
|
|||
public abstract class AbstractMqttV5ClientConnectionTest extends AbstractMqttIntegrationTest { |
|||
|
|||
protected void processClientWithCorrectAccessTokenTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
IMqttToken connectionResult = client.connectAndWait(accessToken); |
|||
MqttWireMessage response = connectionResult.getResponse(); |
|||
|
|||
Assert.assertEquals(MESSAGE_TYPE_CONNACK, response.getType()); |
|||
|
|||
MqttConnAck connAckMsg = (MqttConnAck) response; |
|||
|
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, connAckMsg.getReturnCode()); |
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processClientWithWrongAccessTokenTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
try { |
|||
client.connectAndWait("wrongAccessToken"); |
|||
} catch (MqttException e) { |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_BAD_USERNAME_OR_PASSWORD, e.getReasonCode()); |
|||
} |
|||
} |
|||
|
|||
protected void processClientWithWrongClientIdAndEmptyUsernamePasswordTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient("unknownClientId"); |
|||
try { |
|||
client.connectAndWait(); |
|||
} catch (MqttException e) { |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_IDENTIFIER_NOT_VALID, e.getReasonCode()); |
|||
} |
|||
} |
|||
|
|||
protected void processClientWithNoCredentialsTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(false); |
|||
try { |
|||
client.connectAndWait(); |
|||
} catch (MqttException e) { |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_NOT_AUTHORIZED, e.getReasonCode()); |
|||
} |
|||
} |
|||
|
|||
protected void processClientWithPacketSizeLimitationTest() throws Exception { |
|||
int packetSizeLimit = 99; |
|||
MqttConnectionOptions options = new MqttConnectionOptions(); |
|||
options.setMaximumPacketSize((long) packetSizeLimit); |
|||
options.setUserName(accessToken); |
|||
|
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(options); |
|||
|
|||
MqttV5TestCallback possibleSizeCallback = updateAttributeWithStringValue(client, packetSizeLimit / 2); |
|||
|
|||
Assert.assertTrue("Server should send messages if size less then limitation.", possibleSizeCallback.getPayloadBytes().length < packetSizeLimit); |
|||
|
|||
MqttV5TestCallback bigMessageCallback = updateAttributeWithStringValue(client, packetSizeLimit * 2); |
|||
|
|||
Assert.assertNull("Server should not send a message if the message size bigger then set limit.", bigMessageCallback.getLastReceivedMessage()); |
|||
|
|||
client.disconnect(); |
|||
|
|||
} |
|||
|
|||
private MqttV5TestCallback updateAttributeWithStringValue(MqttV5TestClient client, int valueLen) throws Exception { |
|||
MqttV5TestCallback onUpdateCallback = new MqttV5TestCallback(); |
|||
client.setCallback(onUpdateCallback); |
|||
client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE); |
|||
|
|||
String payload = "{\"sharedStr\":\"" + StringUtils.repeat("*", valueLen) + "\"}"; |
|||
|
|||
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", payload, String.class, status().isOk()); |
|||
onUpdateCallback.getSubscribeLatch().await(3, TimeUnit.SECONDS); |
|||
return onUpdateCallback; |
|||
} |
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.connection; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Ignore; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5ClientConnectionTest extends AbstractMqttV5ClientConnectionTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test MqttV5 client device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithCorrectAccessToken() throws Exception { |
|||
processClientWithCorrectAccessTokenTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithWrongAccessToken() throws Exception { |
|||
processClientWithWrongAccessTokenTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithWrongClientIdAndEmptyUsernamePassword() throws Exception { |
|||
processClientWithWrongClientIdAndEmptyUsernamePasswordTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientWithNoCredentialsTest() throws Exception { |
|||
processClientWithNoCredentialsTest(); |
|||
} |
|||
|
|||
@Test |
|||
@Ignore("Not implemented on the server.") |
|||
public void testClientWithPacketSizeLimitation() throws Exception { |
|||
processClientWithPacketSizeLimitationTest(); |
|||
} |
|||
} |
|||
@ -0,0 +1,77 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.publish; |
|||
|
|||
import org.eclipse.paho.mqttv5.client.IMqttToken; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttPubAck; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage; |
|||
import org.junit.Assert; |
|||
import org.thingsboard.server.common.data.device.profile.MqttTopics; |
|||
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_PUBACK; |
|||
|
|||
public abstract class AbstractMqttV5ClientPublishTest extends AbstractMqttIntegrationTest { |
|||
|
|||
protected static final String PAYLOAD_VALUES_STR = "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4," + |
|||
" \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; |
|||
|
|||
protected static final String INVALID_PAYLOAD_VALUES_STR = "\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4," + |
|||
" \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; |
|||
|
|||
protected void processClientPublishToCorrectTopicTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
IMqttToken publishResult = client.publishAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, PAYLOAD_VALUES_STR.getBytes()); |
|||
MqttWireMessage response = publishResult.getResponse(); |
|||
|
|||
Assert.assertEquals(MESSAGE_TYPE_PUBACK, response.getType()); |
|||
|
|||
MqttPubAck pubAckMsg = (MqttPubAck) response; |
|||
|
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, pubAckMsg.getReturnCode()); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processClientPublishToWrongTopicTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
IMqttToken iMqttToken = client.publishAndWait("wrong/topic/", PAYLOAD_VALUES_STR.getBytes()); |
|||
Assert.assertEquals(MESSAGE_TYPE_PUBACK,iMqttToken.getResponse().getType()); |
|||
MqttPubAck pubAck = (MqttPubAck) iMqttToken.getResponse(); |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_TOPIC_NAME_INVALID, pubAck.getReturnCode()); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processClientPublishWithInvalidPayloadTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
IMqttToken iMqttToken = client.publishAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, INVALID_PAYLOAD_VALUES_STR.getBytes()); |
|||
Assert.assertEquals(MESSAGE_TYPE_PUBACK,iMqttToken.getResponse().getType()); |
|||
MqttPubAck pubAck = (MqttPubAck) iMqttToken.getResponse(); |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_PAYLOAD_FORMAT_INVALID, pubAck.getReturnCode()); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,49 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.publish; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5ClientPublishTest extends AbstractMqttV5ClientPublishTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test MqttV5 client device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientPublishToCorrectTopic() throws Exception { |
|||
processClientPublishToCorrectTopicTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientPublishToWrongTopic() throws Exception { |
|||
processClientPublishToWrongTopicTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientPublishWithInvalidPayload() throws Exception { |
|||
processClientPublishWithInvalidPayloadTest(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,63 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.subscribe; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import org.eclipse.paho.mqttv5.client.IMqttToken; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttSubAck; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage; |
|||
import org.junit.Assert; |
|||
import org.thingsboard.server.common.data.device.profile.MqttTopics; |
|||
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_SUBACK; |
|||
|
|||
public abstract class AbstractMqttV5ClientSubscriptionTest extends AbstractMqttIntegrationTest { |
|||
|
|||
protected void processClientSubscriptionToCorrectTopicTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
IMqttToken subscriptionResult = client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE); |
|||
|
|||
MqttWireMessage response = subscriptionResult.getResponse(); |
|||
|
|||
Assert.assertEquals(MESSAGE_TYPE_SUBACK, response.getType()); |
|||
|
|||
MqttSubAck subAckMsg = (MqttSubAck) response; |
|||
|
|||
Assert.assertEquals(1, subAckMsg.getReturnCodes().length); |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, subAckMsg.getReturnCodes()[0]); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processClientSubscriptionToWrongTopicTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
IMqttToken iMqttToken = client.subscribeAndWait("wrong/topic/+", MqttQoS.AT_MOST_ONCE); |
|||
Assert.assertEquals(MESSAGE_TYPE_SUBACK,iMqttToken.getResponse().getType()); |
|||
MqttSubAck subAck = (MqttSubAck) iMqttToken.getResponse(); |
|||
Assert.assertEquals(1, subAck.getReturnCodes().length); |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_TOPIC_FILTER_NOT_VALID, subAck.getReturnCodes()[0]); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.subscribe; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5ClientSubscriptionTest extends AbstractMqttV5ClientSubscriptionTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test MqttV5 client device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientSubscriptionToCorrectTopic() throws Exception { |
|||
processClientSubscriptionToCorrectTopicTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientSubscriptionToWrongTopic() throws Exception { |
|||
processClientSubscriptionToWrongTopicTest(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.unsubscribe; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import org.eclipse.paho.mqttv5.client.IMqttToken; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttUnsubAck; |
|||
import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage; |
|||
import org.junit.Assert; |
|||
import org.thingsboard.server.common.data.device.profile.MqttTopics; |
|||
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_UNSUBACK; |
|||
|
|||
public abstract class AbstractMqttV5ClientUnsubscribeTest extends AbstractMqttIntegrationTest { |
|||
|
|||
protected void processClientUnsubscribeFromCorrectTopicTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE); |
|||
IMqttToken unsubscribeResult = client.unsubscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC); |
|||
MqttWireMessage response = unsubscribeResult.getResponse(); |
|||
Assert.assertEquals(MESSAGE_TYPE_UNSUBACK, response.getType()); |
|||
MqttUnsubAck unsubAckMsg = (MqttUnsubAck) response; |
|||
Assert.assertEquals(1, unsubAckMsg.getReturnCodes().length); |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, unsubAckMsg.getReturnCodes()[0]); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processClientUnsubscribeWithoutSubscribeTopicTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
|
|||
IMqttToken iMqttToken = client.unsubscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC); |
|||
Assert.assertEquals(MESSAGE_TYPE_UNSUBACK, iMqttToken.getResponse().getType()); |
|||
MqttUnsubAck unsubAck = (MqttUnsubAck) iMqttToken.getResponse(); |
|||
Assert.assertEquals(1, unsubAck.getReturnCodes().length); |
|||
Assert.assertEquals(MqttReturnCode.RETURN_CODE_NO_SUBSCRIPTION_EXISTED, unsubAck.getReturnCodes()[0]); |
|||
|
|||
client.disconnect(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.client.unsubscribe; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5ClientUnsubscribeTest extends AbstractMqttV5ClientUnsubscribeTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test MqttV5 client device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientUnsubscribeFromCorrectTopic() throws Exception { |
|||
processClientUnsubscribeFromCorrectTopicTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testClientUnsubscribeWithoutSubscribeTopic() throws Exception { |
|||
processClientUnsubscribeWithoutSubscribeTopicTest(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,94 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.provision; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.junit.Assert; |
|||
import org.junit.Test; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.DeviceProfileProvisionType; |
|||
import org.thingsboard.server.common.data.TransportPayloadType; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|||
import org.thingsboard.server.dao.device.DeviceCredentialsService; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.dao.device.provision.ProvisionResponseStatus; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_PROVISION_REQUEST_TOPIC; |
|||
import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_PROVISION_RESPONSE_TOPIC; |
|||
|
|||
@Slf4j |
|||
@DaoSqlTest |
|||
public class MqttV5ProvisionDeviceTest extends AbstractMqttV5Test { |
|||
|
|||
@Autowired |
|||
DeviceCredentialsService deviceCredentialsService; |
|||
|
|||
@Autowired |
|||
DeviceService deviceService; |
|||
|
|||
@Test |
|||
public void testProvisioningCreateNewDeviceWithoutCredentials() throws Exception { |
|||
processTestProvisioningCreateNewDeviceWithoutCredentials(); |
|||
} |
|||
|
|||
protected void processTestProvisioningCreateNewDeviceWithoutCredentials() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test Provision device3") |
|||
.transportPayloadType(TransportPayloadType.JSON) |
|||
.provisionType(DeviceProfileProvisionType.ALLOW_CREATE_NEW_DEVICES) |
|||
.provisionKey("testProvisionKey") |
|||
.provisionSecret("testProvisionSecret") |
|||
.build(); |
|||
super.processBeforeTest(configProperties); |
|||
byte[] result = createMqttClientAndPublish(); |
|||
JsonNode response = JacksonUtil.fromBytes(result); |
|||
Assert.assertTrue(response.hasNonNull("credentialsType")); |
|||
Assert.assertTrue(response.hasNonNull("status")); |
|||
|
|||
Device createdDevice = deviceService.findDeviceByTenantIdAndName(tenantId, "Test Provision device"); |
|||
|
|||
Assert.assertNotNull(createdDevice); |
|||
|
|||
DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, createdDevice.getId()); |
|||
|
|||
Assert.assertEquals(deviceCredentials.getCredentialsType().name(), response.get("credentialsType").asText()); |
|||
Assert.assertEquals(ProvisionResponseStatus.SUCCESS.name(), response.get("status").asText()); |
|||
} |
|||
|
|||
protected byte[] createMqttClientAndPublish() throws Exception { |
|||
String provisionRequestMsg = "{\"deviceName\":\"Test Provision device\",\"provisionDeviceKey\":\"testProvisionKey\", \"provisionDeviceSecret\":\"testProvisionSecret\"}"; |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait("provision"); |
|||
MqttV5TestCallback onProvisionCallback = new MqttV5TestCallback(DEVICE_PROVISION_RESPONSE_TOPIC); |
|||
client.setCallback(onProvisionCallback); |
|||
client.subscribe(DEVICE_PROVISION_RESPONSE_TOPIC, MqttQoS.AT_MOST_ONCE); |
|||
client.publishAndWait(DEVICE_PROVISION_REQUEST_TOPIC, provisionRequestMsg.getBytes()); |
|||
onProvisionCallback.getSubscribeLatch().await(3, TimeUnit.SECONDS); |
|||
client.disconnect(); |
|||
return onProvisionCallback.getPayloadBytes(); |
|||
} |
|||
} |
|||
@ -0,0 +1,98 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.rpc; |
|||
|
|||
import com.nimbusds.jose.util.StandardCharset; |
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.eclipse.paho.mqttv5.common.MqttException; |
|||
import org.eclipse.paho.mqttv5.common.MqttMessage; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.junit.Assert.assertEquals; |
|||
import static org.junit.Assert.assertTrue; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractMqttV5RpcTest extends AbstractMqttV5Test { |
|||
|
|||
private static final String DEVICE_RESPONSE = "{\"value1\":\"A\",\"value2\":\"B\"}"; |
|||
|
|||
|
|||
protected void processOneWayRpcTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
MqttV5TestCallback callback = new MqttV5TestCallback(DEVICE_RPC_REQUESTS_SUB_TOPIC.replace("+", "0")); |
|||
client.setCallback(callback); |
|||
client.subscribeAndWait(DEVICE_RPC_REQUESTS_SUB_TOPIC, MqttQoS.AT_MOST_ONCE); |
|||
|
|||
String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"23\",\"value\": 1}}"; |
|||
String result = doPostAsync("/api/rpc/oneway/" + savedDevice.getId(), setGpioRequest, String.class, status().isOk()); |
|||
assertTrue(StringUtils.isEmpty(result)); |
|||
callback.getSubscribeLatch().await(3, TimeUnit.SECONDS); |
|||
assertEquals(JacksonUtil.toJsonNode(setGpioRequest), JacksonUtil.fromBytes(callback.getPayloadBytes())); |
|||
assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS()); |
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected void processJsonTwoWayRpcTest() throws Exception { |
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
client.subscribeAndWait(DEVICE_RPC_REQUESTS_SUB_TOPIC, MqttQoS.AT_LEAST_ONCE); |
|||
MqttV5TestRpcCallback callback = new MqttV5TestRpcCallback(client, DEVICE_RPC_REQUESTS_SUB_TOPIC.replace("+", "0")); |
|||
client.setCallback(callback); |
|||
String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}"; |
|||
String actualRpcResponse = doPostAsync("/api/rpc/twoway/" + savedDevice.getId(), setGpioRequest, String.class, status().isOk()); |
|||
callback.getSubscribeLatch().await(3, TimeUnit.SECONDS); |
|||
assertEquals(JacksonUtil.toJsonNode(setGpioRequest), JacksonUtil.fromBytes(callback.getPayloadBytes())); |
|||
assertEquals("{\"value1\":\"A\",\"value2\":\"B\"}", actualRpcResponse); |
|||
client.disconnect(); |
|||
} |
|||
|
|||
protected class MqttV5TestRpcCallback extends MqttV5TestCallback { |
|||
|
|||
private final MqttV5TestClient client; |
|||
|
|||
public MqttV5TestRpcCallback(MqttV5TestClient client, String awaitSubTopic) { |
|||
super(awaitSubTopic); |
|||
this.client = client; |
|||
} |
|||
|
|||
@Override |
|||
protected void messageArrivedOnAwaitSubTopic(String requestTopic, MqttMessage mqttMessage) { |
|||
log.warn("messageArrived on topic: {}, awaitSubTopic: {}", requestTopic, awaitSubTopic); |
|||
if (awaitSubTopic.equals(requestTopic)) { |
|||
qoS = mqttMessage.getQos(); |
|||
payloadBytes = mqttMessage.getPayload(); |
|||
String responseTopic = requestTopic.replace("request", "response"); |
|||
try { |
|||
client.publish(responseTopic, DEVICE_RESPONSE.getBytes(StandardCharset.UTF_8)); |
|||
} catch (MqttException e) { |
|||
log.warn("Failed to publish response on topic: {} due to: ", responseTopic, e); |
|||
} |
|||
subscribeLatch.countDown(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.rpc; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5RpcTest extends AbstractMqttV5RpcTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("RPC test device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testServerMqttV5SimpleClientOneWayRpc() throws Exception { |
|||
processOneWayRpcTest(); |
|||
} |
|||
|
|||
@Test |
|||
public void testServerMqttV5SimpleClientTwoWayRpc() throws Exception { |
|||
processJsonTwoWayRpcTest(); |
|||
} |
|||
} |
|||
@ -0,0 +1,132 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.timeseries; |
|||
|
|||
import com.fasterxml.jackson.core.type.TypeReference; |
|||
import org.thingsboard.server.common.data.device.profile.MqttTopics; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test; |
|||
import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient; |
|||
|
|||
import java.util.Arrays; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
|
|||
import static org.junit.Assert.assertEquals; |
|||
import static org.junit.Assert.assertNotNull; |
|||
|
|||
public abstract class AbstractMqttV5TimeseriesTest extends AbstractMqttV5Test { |
|||
|
|||
protected static final String PAYLOAD_VALUES_STR = "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4," + |
|||
" \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; |
|||
|
|||
protected void processTimeseriesMqttV5UploadTest() throws Exception { |
|||
|
|||
List<String> expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); |
|||
|
|||
MqttV5TestClient client = new MqttV5TestClient(); |
|||
client.connectAndWait(accessToken); |
|||
client.publishAndWait(MqttTopics.DEVICE_TELEMETRY_TOPIC, PAYLOAD_VALUES_STR.getBytes()); |
|||
client.disconnect(); |
|||
|
|||
DeviceId deviceId = savedDevice.getId(); |
|||
|
|||
List<String> actualKeys = getActualKeysList(deviceId, expectedKeys); |
|||
assertNotNull(actualKeys); |
|||
|
|||
Set<String> actualKeySet = new HashSet<>(actualKeys); |
|||
Set<String> expectedKeySet = new HashSet<>(expectedKeys); |
|||
|
|||
assertEquals(expectedKeySet, actualKeySet); |
|||
|
|||
String getTelemetryValuesUrl; |
|||
getTelemetryValuesUrl = "/api/plugins/telemetry/DEVICE/" + deviceId + "/values/timeseries?keys=" + String.join(",", actualKeySet); |
|||
long start = System.currentTimeMillis(); |
|||
long end = System.currentTimeMillis() + 5000; |
|||
Map<String, List<Map<String, Object>>> values = null; |
|||
while (start <= end) { |
|||
values = doGetAsyncTyped(getTelemetryValuesUrl, new TypeReference<>() { |
|||
}); |
|||
boolean valid = values.size() == expectedKeys.size(); |
|||
if (valid) { |
|||
for (String key : expectedKeys) { |
|||
List<Map<String, Object>> tsValues = values.get(key); |
|||
if (tsValues != null && tsValues.size() > 0) { |
|||
Object ts = tsValues.get(0).get("ts"); |
|||
if (ts == null) { |
|||
valid = false; |
|||
break; |
|||
} |
|||
} else { |
|||
valid = false; |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
if (valid) { |
|||
break; |
|||
} |
|||
Thread.sleep(100); |
|||
start += 100; |
|||
} |
|||
assertNotNull(values); |
|||
assertValues(values); |
|||
} |
|||
|
|||
private List<String> getActualKeysList(DeviceId deviceId, List<String> expectedKeys) throws Exception { |
|||
long start = System.currentTimeMillis(); |
|||
long end = System.currentTimeMillis() + 3000; |
|||
|
|||
List<String> actualKeys = null; |
|||
while (start <= end) { |
|||
actualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/timeseries", new TypeReference<>() { |
|||
}); |
|||
if (actualKeys.size() == expectedKeys.size()) { |
|||
break; |
|||
} |
|||
Thread.sleep(100); |
|||
start += 100; |
|||
} |
|||
return actualKeys; |
|||
} |
|||
|
|||
private void assertValues(Map<String, List<Map<String, Object>>> deviceValues) { |
|||
for (Map.Entry<String, List<Map<String, Object>>> entry : deviceValues.entrySet()) { |
|||
String key = entry.getKey(); |
|||
List<Map<String, Object>> tsKv = entry.getValue(); |
|||
String value = (String) tsKv.get(0).get("value"); |
|||
switch (key) { |
|||
case "key1": |
|||
assertEquals("value1", value); |
|||
break; |
|||
case "key2": |
|||
assertEquals("true", value); |
|||
break; |
|||
case "key3": |
|||
assertEquals("3.0", value); |
|||
break; |
|||
case "key4": |
|||
assertEquals("4", value); |
|||
break; |
|||
case "key5": |
|||
assertEquals("{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}", value); |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.mqttv5.timeseries; |
|||
|
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
@DaoSqlTest |
|||
public class MqttV5TimeseriesTest extends AbstractMqttV5TimeseriesTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("Test Post Telemetry device") |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testTimeseriesMqttV5SimpleClientUpload() throws Exception { |
|||
processTimeseriesMqttV5UploadTest(); |
|||
} |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue