简柏林 / thingskit · Commits

Commit fb098e1bae920b89358596ee274f4fd960673952

Authored by Viacheslav Kukhtyn
1 parent 67b8c2f4

Cover MQTT API with black box tests

Inline Side-by-side
Showing 10 changed files with 484 additions and 57 deletions
... ... @@ -15,4 +15,4 @@ As result, in REPOSITORY column, next images should be present:
15 15
16 16 - Run the integration tests in the [msa/integration-tests](../integration-tests) directory:
17 17
18   - mvn clean install -Dintegrationtests.skip=false
\ No newline at end of file
  18 + mvn clean install -DintegrationTests.skip=false
\ No newline at end of file
... ...
... ... @@ -34,9 +34,10 @@
34 34 <properties>
35 35 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
36 36 <main.dir>${basedir}/../..</main.dir>
37   - <integrationtests.skip>true</integrationtests.skip>
  37 + <integrationTests.skip>true</integrationTests.skip>
38 38 <testcontainers.version>1.9.1</testcontainers.version>
39 39 <java-websocket.version>1.3.9</java-websocket.version>
  40 + <httpclient.version>4.5.6</httpclient.version>
40 41 </properties>
41 42
42 43 <dependencies>
... ... @@ -51,6 +52,11 @@
51 52 <version>${java-websocket.version}</version>
52 53 </dependency>
53 54 <dependency>
  55 + <groupId>org.apache.httpcomponents</groupId>
  56 + <artifactId>httpclient</artifactId>
  57 + <version>${httpclient.version}</version>
  58 + </dependency>
  59 + <dependency>
54 60 <groupId>io.takari.junit</groupId>
55 61 <artifactId>takari-cpsuite</artifactId>
56 62 </dependency>
... ... @@ -89,7 +95,7 @@
89 95 <includes>
90 96 <include>**/*TestSuite.java</include>
91 97 </includes>
92   - <skipTests>${integrationtests.skip}</skipTests>
  98 + <skipTests>${integrationTests.skip}</skipTests>
93 99 </configuration>
94 100 </plugin>
95 101 </plugins>
... ...
... ... @@ -21,55 +21,68 @@ import com.google.gson.JsonArray;
21 21 import com.google.gson.JsonObject;
22 22 import lombok.extern.slf4j.Slf4j;
23 23 import org.apache.commons.lang3.RandomStringUtils;
  24 +import org.apache.http.config.Registry;
  25 +import org.apache.http.config.RegistryBuilder;
  26 +import org.apache.http.conn.socket.ConnectionSocketFactory;
  27 +import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
  28 +import org.apache.http.conn.ssl.TrustStrategy;
  29 +import org.apache.http.conn.ssl.X509HostnameVerifier;
  30 +import org.apache.http.impl.client.CloseableHttpClient;
  31 +import org.apache.http.impl.client.HttpClients;
  32 +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
  33 +import org.apache.http.ssl.SSLContextBuilder;
  34 +import org.apache.http.ssl.SSLContexts;
24 35 import org.junit.*;
  36 +import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
25 37 import org.thingsboard.client.tools.RestClient;
26 38 import org.thingsboard.server.common.data.Device;
27 39 import org.thingsboard.server.common.data.EntityType;
28 40 import org.thingsboard.server.common.data.id.DeviceId;
  41 +import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
29 42
  43 +import javax.net.ssl.*;
30 44 import java.net.URI;
31   -import java.net.URISyntaxException;
  45 +import java.security.cert.X509Certificate;
32 46 import java.util.List;
33 47 import java.util.Map;
34 48 import java.util.Random;
35   -import java.util.concurrent.TimeUnit;
36 49
37 50 @Slf4j
38 51 public abstract class AbstractContainerTest {
39   - protected static String httpUrl;
40   - protected static String wsUrl;
  52 + protected static final String HTTPS_URL = "https://localhost";
  53 + protected static final String WSS_URL = "wss://localhost";
41 54 protected static RestClient restClient;
42 55 protected ObjectMapper mapper = new ObjectMapper();
43 56
44 57 @BeforeClass
45   - public static void before() {
46   - httpUrl = "http://localhost:" + ContainerTestSuite.composeContainer.getServicePort("tb-web-ui1", ContainerTestSuite.EXPOSED_PORT);
47   - wsUrl = "ws://localhost:" + ContainerTestSuite.composeContainer.getServicePort("tb-web-ui1", ContainerTestSuite.EXPOSED_PORT);
48   - restClient = new RestClient(httpUrl);
  58 + public static void before() throws Exception {
  59 + restClient = new RestClient(HTTPS_URL);
  60 + restClient.getRestTemplate().setRequestFactory(getRequestFactoryForSelfSignedCert());
49 61 }
50 62
51 63 protected Device createDevice(String name) {
52 64 return restClient.createDevice(name + RandomStringUtils.randomAlphanumeric(7), "DEFAULT");
53 65 }
54 66
55   - protected WsClient subscribeToTelemetryWebSocket(DeviceId deviceId) throws URISyntaxException, InterruptedException {
56   - WsClient mWs = new WsClient(new URI(wsUrl + "/api/ws/plugins/telemetry?token=" + restClient.getToken()));
57   - mWs.connectBlocking(1, TimeUnit.SECONDS);
58   -
59   - JsonObject tsSubCmd = new JsonObject();
60   - tsSubCmd.addProperty("entityType", EntityType.DEVICE.name());
61   - tsSubCmd.addProperty("entityId", deviceId.toString());
62   - tsSubCmd.addProperty("scope", "LATEST_TELEMETRY");
63   - tsSubCmd.addProperty("cmdId", new Random().nextInt(100));
64   - tsSubCmd.addProperty("unsubscribe", false);
65   - JsonArray wsTsSubCmds = new JsonArray();
66   - wsTsSubCmds.add(tsSubCmd);
  67 + protected WsClient subscribeToWebSocket(DeviceId deviceId, String scope, CmdsType property) throws Exception {
  68 + WsClient wsClient = new WsClient(new URI(WSS_URL + "/api/ws/plugins/telemetry?token=" + restClient.getToken()));
  69 + SSLContextBuilder builder = SSLContexts.custom();
  70 + builder.loadTrustMaterial(null, (TrustStrategy) (chain, authType) -> true);
  71 + wsClient.setSocket(builder.build().getSocketFactory().createSocket());
  72 + wsClient.connectBlocking();
  73 +
  74 + JsonObject cmdsObject = new JsonObject();
  75 + cmdsObject.addProperty("entityType", EntityType.DEVICE.name());
  76 + cmdsObject.addProperty("entityId", deviceId.toString());
  77 + cmdsObject.addProperty("scope", scope);
  78 + cmdsObject.addProperty("cmdId", new Random().nextInt(100));
  79 +
  80 + JsonArray cmd = new JsonArray();
  81 + cmd.add(cmdsObject);
67 82 JsonObject wsRequest = new JsonObject();
68   - wsRequest.add("tsSubCmds", wsTsSubCmds);
69   - wsRequest.add("historyCmds", new JsonArray());
70   - wsRequest.add("attrSubCmds", new JsonArray());
71   - mWs.send(wsRequest.toString());
72   - return mWs;
  83 + wsRequest.add(property.toString(), cmd);
  84 + wsClient.send(wsRequest.toString());
  85 + return wsClient;
73 86 }
74 87
75 88 protected Map<String, Long> getExpectedLatestValues(long ts) {
... ... @@ -109,4 +122,54 @@ public abstract class AbstractContainerTest {
109 122 return values;
110 123 }
111 124
  125 + protected enum CmdsType {
  126 + TS_SUB_CMDS("tsSubCmds"),
  127 + HISTORY_CMDS("historyCmds"),
  128 + ATTR_SUB_CMDS("attrSubCmds");
  129 +
  130 + private final String text;
  131 +
  132 + CmdsType(final String text) {
  133 + this.text = text;
  134 + }
  135 +
  136 + @Override
  137 + public String toString() {
  138 + return text;
  139 + }
  140 + }
  141 +
  142 + private static HttpComponentsClientHttpRequestFactory getRequestFactoryForSelfSignedCert() throws Exception {
  143 + SSLContextBuilder builder = SSLContexts.custom();
  144 + builder.loadTrustMaterial(null, (TrustStrategy) (chain, authType) -> true);
  145 + SSLContext sslContext = builder.build();
  146 + SSLConnectionSocketFactory sslSelfSigned = new SSLConnectionSocketFactory(sslContext, new X509HostnameVerifier() {
  147 + @Override
  148 + public void verify(String host, SSLSocket ssl) {
  149 + }
  150 +
  151 + @Override
  152 + public void verify(String host, X509Certificate cert) {
  153 + }
  154 +
  155 + @Override
  156 + public void verify(String host, String[] cns, String[] subjectAlts) {
  157 + }
  158 +
  159 + @Override
  160 + public boolean verify(String s, SSLSession sslSession) {
  161 + return true;
  162 + }
  163 + });
  164 +
  165 + Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder
  166 + .<ConnectionSocketFactory>create()
  167 + .register("https", sslSelfSigned)
  168 + .build();
  169 +
  170 + PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(socketFactoryRegistry);
  171 + CloseableHttpClient httpClient = HttpClients.custom().setConnectionManager(cm).build();
  172 + return new HttpComponentsClientHttpRequestFactory(httpClient);
  173 + }
  174 +
112 175 }
... ...
... ... @@ -26,12 +26,11 @@ import java.io.File;
26 26 @RunWith(ClasspathSuite.class)
27 27 @ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*"})
28 28 public class ContainerTestSuite {
29   - static final int EXPOSED_PORT = 8080;
30 29
31 30 @ClassRule
32 31 public static DockerComposeContainer composeContainer = new DockerComposeContainer(new File("./../docker/docker-compose.yml"))
33 32 .withPull(false)
34 33 .withLocalCompose(true)
35 34 .withTailChildContainers(true)
36   - .withExposedService("tb-web-ui1", EXPOSED_PORT, Wait.forHttp("/login"));
  35 + .withExposedService("tb-web-ui1", 8080, Wait.forHttp("/login"));
37 36 }
... ...
... ... @@ -19,16 +19,12 @@ import org.java_websocket.client.WebSocketClient;
19 19 import org.java_websocket.handshake.ServerHandshake;
20 20
21 21 import java.net.URI;
22   -import java.util.concurrent.ArrayBlockingQueue;
23   -import java.util.concurrent.BlockingQueue;
24 22
25 23 public class WsClient extends WebSocketClient {
26   - private final BlockingQueue<String> events;
27 24 private String message;
28 25
29 26 public WsClient(URI serverUri) {
30 27 super(serverUri);
31   - events = new ArrayBlockingQueue<>(100);
32 28 }
33 29
34 30 @Override
... ... @@ -37,13 +33,11 @@ public class WsClient extends WebSocketClient {
37 33
38 34 @Override
39 35 public void onMessage(String message) {
40   - events.add(message);
41 36 this.message = message;
42 37 }
43 38
44 39 @Override
45 40 public void onClose(int code, String reason, boolean remote) {
46   - events.clear();
47 41 }
48 42
49 43 @Override
... ... @@ -54,4 +48,4 @@ public class WsClient extends WebSocketClient {
54 48 public String getLastMessage() {
55 49 return this.message;
56 50 }
57   -}
\ No newline at end of file
  51 +}
... ...
... ... @@ -15,6 +15,7 @@
15 15 */
16 16 package org.thingsboard.server.msa.connectivity;
17 17
  18 +import com.google.common.collect.Sets;
18 19 import org.junit.Assert;
19 20 import org.junit.Test;
20 21 import org.springframework.http.ResponseEntity;
... ... @@ -22,7 +23,7 @@ import org.thingsboard.server.common.data.Device;
22 23 import org.thingsboard.server.common.data.security.DeviceCredentials;
23 24 import org.thingsboard.server.msa.AbstractContainerTest;
24 25 import org.thingsboard.server.msa.WsClient;
25   -import org.thingsboard.server.msa.WsTelemetryResponse;
  26 +import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
26 27
27 28 import java.util.concurrent.TimeUnit;
28 29
... ... @@ -35,23 +36,24 @@ public class HttpClientTest extends AbstractContainerTest {
35 36 Device device = createDevice("http_");
36 37 DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
37 38
38   - WsClient mWs = subscribeToTelemetryWebSocket(device.getId());
  39 + WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS);
39 40 ResponseEntity deviceTelemetryResponse = restClient.getRestTemplate()
40   - .postForEntity(httpUrl + "/api/v1/{credentialsId}/telemetry",
  41 + .postForEntity(HTTPS_URL + "/api/v1/{credentialsId}/telemetry",
41 42 mapper.readTree(createPayload().toString()),
42 43 ResponseEntity.class,
43 44 deviceCredentials.getCredentialsId());
44 45 Assert.assertTrue(deviceTelemetryResponse.getStatusCode().is2xxSuccessful());
45 46 TimeUnit.SECONDS.sleep(1);
46   - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class);
  47 + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class);
47 48
48   - Assert.assertEquals(getExpectedLatestValues(123456789L).keySet(), actualLatestTelemetry.getLatestValues().keySet());
  49 + Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"),
  50 + actualLatestTelemetry.getLatestValues().keySet());
49 51
50 52 Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString()));
51 53 Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1"));
52 54 Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0)));
53 55 Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73)));
54 56
55   - restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId());
  57 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
56 58 }
57 59 }
... ...
... ... @@ -15,20 +15,39 @@
15 15 */
16 16 package org.thingsboard.server.msa.connectivity;
17 17
  18 +import com.fasterxml.jackson.databind.JsonNode;
  19 +import com.google.common.collect.Sets;
  20 +import com.google.common.util.concurrent.ListenableFuture;
  21 +import com.google.common.util.concurrent.ListeningExecutorService;
  22 +import com.google.common.util.concurrent.MoreExecutors;
  23 +import com.google.gson.JsonObject;
18 24 import io.netty.buffer.ByteBuf;
19 25 import io.netty.buffer.Unpooled;
20 26 import lombok.Data;
  27 +import org.apache.commons.lang3.RandomStringUtils;
21 28 import org.junit.*;
  29 +import org.springframework.core.ParameterizedTypeReference;
  30 +import org.springframework.http.HttpMethod;
  31 +import org.springframework.http.ResponseEntity;
22 32 import org.thingsboard.mqtt.MqttClient;
23 33 import org.thingsboard.mqtt.MqttClientConfig;
24 34 import org.thingsboard.mqtt.MqttHandler;
25 35 import org.thingsboard.server.common.data.Device;
  36 +import org.thingsboard.server.common.data.id.RuleChainId;
  37 +import org.thingsboard.server.common.data.page.TextPageData;
  38 +import org.thingsboard.server.common.data.rule.NodeConnectionInfo;
  39 +import org.thingsboard.server.common.data.rule.RuleChain;
  40 +import org.thingsboard.server.common.data.rule.RuleChainMetaData;
  41 +import org.thingsboard.server.common.data.rule.RuleNode;
26 42 import org.thingsboard.server.common.data.security.DeviceCredentials;
27 43 import org.thingsboard.server.msa.AbstractContainerTest;
28 44 import org.thingsboard.server.msa.WsClient;
29   -import org.thingsboard.server.msa.WsTelemetryResponse;
  45 +import org.thingsboard.server.msa.mapper.AttributesResponse;
  46 +import org.thingsboard.server.msa.mapper.WsTelemetryResponse;
30 47
  48 +import java.io.IOException;
31 49 import java.nio.charset.StandardCharsets;
  50 +import java.util.*;
32 51 import java.util.concurrent.*;
33 52
34 53 public class MqttClientTest extends AbstractContainerTest {
... ... @@ -39,20 +58,22 @@ public class MqttClientTest extends AbstractContainerTest {
39 58 Device device = createDevice("mqtt_");
40 59 DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
41 60
42   - WsClient mWs = subscribeToTelemetryWebSocket(device.getId());
43   - MqttClient mqttClient = getMqttClient(deviceCredentials);
  61 + WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS);
  62 + MqttClient mqttClient = getMqttClient(deviceCredentials, null);
44 63 mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload().toString().getBytes()));
45 64 TimeUnit.SECONDS.sleep(1);
46   - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class);
  65 + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class);
47 66
48   - Assert.assertEquals(getExpectedLatestValues(123456789L).keySet(), actualLatestTelemetry.getLatestValues().keySet());
  67 + Assert.assertEquals(4, actualLatestTelemetry.getData().size());
  68 + Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"),
  69 + actualLatestTelemetry.getLatestValues().keySet());
49 70
50 71 Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString()));
51 72 Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1"));
52 73 Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0)));
53 74 Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73)));
54 75
55   - restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId());
  76 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
56 77 }
57 78
58 79 @Test
... ... @@ -63,12 +84,13 @@ public class MqttClientTest extends AbstractContainerTest {
63 84 Device device = createDevice("mqtt_");
64 85 DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
65 86
66   - WsClient mWs = subscribeToTelemetryWebSocket(device.getId());
67   - MqttClient mqttClient = getMqttClient(deviceCredentials);
  87 + WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS);
  88 + MqttClient mqttClient = getMqttClient(deviceCredentials, null);
68 89 mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload(ts).toString().getBytes()));
69 90 TimeUnit.SECONDS.sleep(1);
70   - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class);
  91 + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class);
71 92
  93 + Assert.assertEquals(4, actualLatestTelemetry.getData().size());
72 94 Assert.assertEquals(getExpectedLatestValues(ts), actualLatestTelemetry.getLatestValues());
73 95
74 96 Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", ts, Boolean.TRUE.toString()));
... ... @@ -76,15 +98,271 @@ public class MqttClientTest extends AbstractContainerTest {
76 98 Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", ts, Double.toString(42.0)));
77 99 Assert.assertTrue(verify(actualLatestTelemetry, "longKey", ts, Long.toString(73)));
78 100
79   - restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId());
  101 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
80 102 }
81 103
82   - private MqttClient getMqttClient(DeviceCredentials deviceCredentials) throws InterruptedException {
83   - MqttMessageListener queue = new MqttMessageListener();
  104 + @Test
  105 + public void publishAttributeUpdateToServer() throws Exception {
  106 + restClient.login("tenant@thingsboard.org", "tenant");
  107 + Device device = createDevice("mqtt_");
  108 + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
  109 +
  110 + WsClient wsClient = subscribeToWebSocket(device.getId(), "CLIENT_SCOPE", CmdsType.ATTR_SUB_CMDS);
  111 + MqttMessageListener listener = new MqttMessageListener();
  112 + MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
  113 + JsonObject clientAttributes = new JsonObject();
  114 + clientAttributes.addProperty("attr1", "value1");
  115 + clientAttributes.addProperty("attr2", true);
  116 + clientAttributes.addProperty("attr3", 42.0);
  117 + clientAttributes.addProperty("attr4", 73);
  118 + mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes()));
  119 + TimeUnit.SECONDS.sleep(1);
  120 + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class);
  121 +
  122 + Assert.assertEquals(4, actualLatestTelemetry.getData().size());
  123 + Assert.assertEquals(Sets.newHashSet("attr1", "attr2", "attr3", "attr4"),
  124 + actualLatestTelemetry.getLatestValues().keySet());
  125 +
  126 + Assert.assertTrue(verify(actualLatestTelemetry, "attr1", "value1"));
  127 + Assert.assertTrue(verify(actualLatestTelemetry, "attr2", Boolean.TRUE.toString()));
  128 + Assert.assertTrue(verify(actualLatestTelemetry, "attr3", Double.toString(42.0)));
  129 + Assert.assertTrue(verify(actualLatestTelemetry, "attr4", Long.toString(73)));
  130 +
  131 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
  132 + }
  133 +
  134 + @Test
  135 + public void requestAttributeValuesFromServer() throws Exception {
  136 + restClient.login("tenant@thingsboard.org", "tenant");
  137 + Device device = createDevice("mqtt_");
  138 + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
  139 +
  140 + MqttMessageListener listener = new MqttMessageListener();
  141 + MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
  142 +
  143 + // Add a new client attribute
  144 + JsonObject clientAttributes = new JsonObject();
  145 + String clientAttributeValue = RandomStringUtils.randomAlphanumeric(8);
  146 + clientAttributes.addProperty("clientAttr", clientAttributeValue);
  147 + mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes()));
  148 +
  149 + // Add a new shared attribute
  150 + JsonObject sharedAttributes = new JsonObject();
  151 + String sharedAttributeValue = RandomStringUtils.randomAlphanumeric(8);
  152 + sharedAttributes.addProperty("sharedAttr", sharedAttributeValue);
  153 + ResponseEntity sharedAttributesResponse = restClient.getRestTemplate()
  154 + .postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
  155 + mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
  156 + device.getId());
  157 + Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
  158 +
  159 + // Subscribe to attributes response
  160 + mqttClient.on("v1/devices/me/attributes/response/+", listener);
  161 + // Request attributes
  162 + JsonObject request = new JsonObject();
  163 + request.addProperty("clientKeys", "clientAttr");
  164 + request.addProperty("sharedKeys", "sharedAttr");
  165 + mqttClient.publish("v1/devices/me/attributes/request/" + new Random().nextInt(100), Unpooled.wrappedBuffer(request.toString().getBytes()));
  166 + MqttEvent event = listener.getEvents().poll(10, TimeUnit.SECONDS);
  167 + AttributesResponse attributes = mapper.readValue(Objects.requireNonNull(event).getMessage(), AttributesResponse.class);
  168 +
  169 + Assert.assertEquals(1, attributes.getClient().size());
  170 + Assert.assertEquals(clientAttributeValue, attributes.getClient().get("clientAttr"));
  171 +
  172 + Assert.assertEquals(1, attributes.getShared().size());
  173 + Assert.assertEquals(sharedAttributeValue, attributes.getShared().get("sharedAttr"));
  174 +
  175 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
  176 + }
  177 +
  178 + @Test
  179 + public void subscribeToAttributeUpdatesFromServer() throws Exception {
  180 + restClient.login("tenant@thingsboard.org", "tenant");
  181 + Device device = createDevice("mqtt_");
  182 + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
  183 +
  184 + MqttMessageListener listener = new MqttMessageListener();
  185 + MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
  186 + mqttClient.on("v1/devices/me/attributes", listener);
  187 +
  188 + String sharedAttributeName = "sharedAttr";
  189 +
  190 + // Add a new shared attribute
  191 + JsonObject sharedAttributes = new JsonObject();
  192 + String sharedAttributeValue = RandomStringUtils.randomAlphanumeric(8);
  193 + sharedAttributes.addProperty(sharedAttributeName, sharedAttributeValue);
  194 + ResponseEntity sharedAttributesResponse = restClient.getRestTemplate()
  195 + .postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
  196 + mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
  197 + device.getId());
  198 + Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
  199 +
  200 + MqttEvent event = listener.getEvents().poll(10, TimeUnit.SECONDS);
  201 + Assert.assertEquals(sharedAttributeValue,
  202 + mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText());
  203 +
  204 + // Update the shared attribute value
  205 + JsonObject updatedSharedAttributes = new JsonObject();
  206 + String updatedSharedAttributeValue = RandomStringUtils.randomAlphanumeric(8);
  207 + updatedSharedAttributes.addProperty(sharedAttributeName, updatedSharedAttributeValue);
  208 + ResponseEntity updatedSharedAttributesResponse = restClient.getRestTemplate()
  209 + .postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
  210 + mapper.readTree(updatedSharedAttributes.toString()), ResponseEntity.class,
  211 + device.getId());
  212 + Assert.assertTrue(updatedSharedAttributesResponse.getStatusCode().is2xxSuccessful());
  213 +
  214 + event = listener.getEvents().poll(10, TimeUnit.SECONDS);
  215 + Assert.assertEquals(updatedSharedAttributeValue,
  216 + mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText());
  217 +
  218 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
  219 + }
  220 +
  221 + @Test
  222 + public void serverSideRpc() throws Exception {
  223 + restClient.login("tenant@thingsboard.org", "tenant");
  224 + Device device = createDevice("mqtt_");
  225 + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
  226 +
  227 + MqttMessageListener listener = new MqttMessageListener();
  228 + MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
  229 + mqttClient.on("v1/devices/me/rpc/request/+", listener);
  230 +
  231 + // Send an RPC from the server
  232 + JsonObject serverRpcPayload = new JsonObject();
  233 + serverRpcPayload.addProperty("method", "getValue");
  234 + serverRpcPayload.addProperty("params", true);
  235 + serverRpcPayload.addProperty("timeout", 1000);
  236 + ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
  237 + ListenableFuture<ResponseEntity> future = service.submit(() -> {
  238 + try {
  239 + return restClient.getRestTemplate()
  240 + .postForEntity(HTTPS_URL + "/api/plugins/rpc/twoway/{deviceId}",
  241 + mapper.readTree(serverRpcPayload.toString()), String.class,
  242 + device.getId());
  243 + } catch (IOException e) {
  244 + return ResponseEntity.badRequest().build();
  245 + }
  246 + });
  247 +
  248 + // Wait for RPC call from the server and send the response
  249 + MqttEvent requestFromServer = listener.getEvents().poll(10, TimeUnit.SECONDS);
  250 +
  251 + Assert.assertEquals("{\"method\":\"getValue\",\"params\":true}", Objects.requireNonNull(requestFromServer).getMessage());
  252 +
  253 + Integer requestId = Integer.valueOf(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length()));
  254 + JsonObject clientResponse = new JsonObject();
  255 + clientResponse.addProperty("response", "someResponse");
  256 + // Send a response to the server's RPC request
  257 + mqttClient.publish("v1/devices/me/rpc/response/" + requestId, Unpooled.wrappedBuffer(clientResponse.toString().getBytes()));
  258 +
  259 + ResponseEntity serverResponse = future.get(5, TimeUnit.SECONDS);
  260 + Assert.assertTrue(serverResponse.getStatusCode().is2xxSuccessful());
  261 + Assert.assertEquals(clientResponse.toString(), serverResponse.getBody());
  262 +
  263 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
  264 + }
  265 +
  266 + @Test
  267 + public void clientSideRpc() throws Exception {
  268 + restClient.login("tenant@thingsboard.org", "tenant");
  269 + Device device = createDevice("mqtt_");
  270 + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId());
  271 +
  272 + MqttMessageListener listener = new MqttMessageListener();
  273 + MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
  274 + mqttClient.on("v1/devices/me/rpc/request/+", listener);
  275 +
  276 + // Get the default rule chain id to make it root again after test finished
  277 + RuleChainId defaultRuleChainId = getDefaultRuleChainId();
  278 +
  279 + // Create a new root rule chain
  280 + RuleChainId ruleChainId = createRootRuleChainForRpcResponse();
  281 +
  282 + // Send the request to the server
  283 + JsonObject clientRequest = new JsonObject();
  284 + clientRequest.addProperty("method", "getResponse");
  285 + clientRequest.addProperty("params", true);
  286 + Integer requestId = 42;
  287 + mqttClient.publish("v1/devices/me/rpc/request/" + requestId, Unpooled.wrappedBuffer(clientRequest.toString().getBytes()));
  288 +
  289 + // Check the response from the server
  290 + TimeUnit.SECONDS.sleep(1);
  291 + MqttEvent responseFromServer = listener.getEvents().poll(1, TimeUnit.SECONDS);
  292 + Integer responseId = Integer.valueOf(Objects.requireNonNull(responseFromServer).getTopic().substring("v1/devices/me/rpc/response/".length()));
  293 + Assert.assertEquals(requestId, responseId);
  294 + Assert.assertEquals("requestReceived", mapper.readTree(responseFromServer.getMessage()).get("response").asText());
  295 +
  296 + // Make the default rule chain a root again
  297 + ResponseEntity<RuleChain> rootRuleChainResponse = restClient.getRestTemplate()
  298 + .postForEntity(HTTPS_URL + "/api/ruleChain/{ruleChainId}/root",
  299 + null,
  300 + RuleChain.class,
  301 + defaultRuleChainId);
  302 + Assert.assertTrue(rootRuleChainResponse.getStatusCode().is2xxSuccessful());
  303 +
  304 + // Delete the created rule chain
  305 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/ruleChain/{ruleChainId}", ruleChainId);
  306 + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
  307 + }
  308 +
  309 + private RuleChainId createRootRuleChainForRpcResponse() throws Exception {
  310 + RuleChain newRuleChain = new RuleChain();
  311 + newRuleChain.setName("testRuleChain");
  312 + ResponseEntity<RuleChain> ruleChainResponse = restClient.getRestTemplate()
  313 + .postForEntity(HTTPS_URL + "/api/ruleChain",
  314 + newRuleChain,
  315 + RuleChain.class);
  316 + Assert.assertTrue(ruleChainResponse.getStatusCode().is2xxSuccessful());
  317 + RuleChain ruleChain = ruleChainResponse.getBody();
  318 +
  319 + JsonNode configuration = mapper.readTree(this.getClass().getClassLoader().getResourceAsStream("RpcResponseRuleChainMetadata.json"));
  320 + RuleChainMetaData ruleChainMetaData = new RuleChainMetaData();
  321 + ruleChainMetaData.setRuleChainId(ruleChain.getId());
  322 + ruleChainMetaData.setFirstNodeIndex(configuration.get("firstNodeIndex").asInt());
  323 + ruleChainMetaData.setNodes(Arrays.asList(mapper.treeToValue(configuration.get("nodes"), RuleNode[].class)));
  324 + ruleChainMetaData.setConnections(Arrays.asList(mapper.treeToValue(configuration.get("connections"), NodeConnectionInfo[].class)));
  325 +
  326 + ResponseEntity<RuleChainMetaData> ruleChainMetadataResponse = restClient.getRestTemplate()
  327 + .postForEntity(HTTPS_URL + "/api/ruleChain/metadata",
  328 + ruleChainMetaData,
  329 + RuleChainMetaData.class);
  330 + Assert.assertTrue(ruleChainMetadataResponse.getStatusCode().is2xxSuccessful());
  331 +
  332 + // Set a new rule chain as root
  333 + ResponseEntity<RuleChain> rootRuleChainResponse = restClient.getRestTemplate()
  334 + .postForEntity(HTTPS_URL + "/api/ruleChain/{ruleChainId}/root",
  335 + null,
  336 + RuleChain.class,
  337 + ruleChain.getId());
  338 + Assert.assertTrue(rootRuleChainResponse.getStatusCode().is2xxSuccessful());
  339 +
  340 + return ruleChain.getId();
  341 + }
  342 +
  343 + private RuleChainId getDefaultRuleChainId() {
  344 + ResponseEntity<TextPageData<RuleChain>> ruleChains = restClient.getRestTemplate().exchange(
  345 + HTTPS_URL + "/api/ruleChains?limit=40&textSearch=",
  346 + HttpMethod.GET,
  347 + null,
  348 + new ParameterizedTypeReference<TextPageData<RuleChain>>() {
  349 + });
  350 +
  351 + Optional<RuleChain> defaultRuleChain = ruleChains.getBody().getData()
  352 + .stream()
  353 + .filter(RuleChain::isRoot)
  354 + .findFirst();
  355 + if (!defaultRuleChain.isPresent()) {
  356 + Assert.fail("Root rule chain wasn't found");
  357 + }
  358 + return defaultRuleChain.get().getId();
  359 + }
  360 +
  361 + private MqttClient getMqttClient(DeviceCredentials deviceCredentials, MqttMessageListener listener) throws InterruptedException {
84 362 MqttClientConfig clientConfig = new MqttClientConfig();
85 363 clientConfig.setClientId("MQTT client from test");
86 364 clientConfig.setUsername(deviceCredentials.getCredentialsId());
87   - MqttClient mqttClient = MqttClient.create(clientConfig, queue);
  365 + MqttClient mqttClient = MqttClient.create(clientConfig, listener);
88 366 mqttClient.connect("localhost", 1883).sync();
89 367 return mqttClient;
90 368 }
... ...
  1 +/**
  2 + * Copyright © 2016-2018 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.msa.mapper;
  17 +
  18 +import lombok.Data;
  19 +
  20 +import java.util.Map;
  21 +
  22 +@Data
  23 +public class AttributesResponse {
  24 + private Map<String, Object> client;
  25 + private Map<String, Object> shared;
  26 +}
... ...
msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java renamed from msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java
... ... @@ -13,7 +13,7 @@
13 13 * See the License for the specific language governing permissions and
14 14 * limitations under the License.
15 15 */
16   -package org.thingsboard.server.msa;
  16 +package org.thingsboard.server.msa.mapper;
17 17
18 18 import lombok.Data;
19 19
... ...
  1 +{
  2 + "firstNodeIndex": 0,
  3 + "nodes": [
  4 + {
  5 + "additionalInfo": {
  6 + "layoutX": 325,
  7 + "layoutY": 150
  8 + },
  9 + "type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode",
  10 + "name": "msgTypeSwitch",
  11 + "debugMode": true,
  12 + "configuration": {
  13 + "version": 0
  14 + }
  15 + },
  16 + {
  17 + "additionalInfo": {
  18 + "layoutX": 60,
  19 + "layoutY": 300
  20 + },
  21 + "type": "org.thingsboard.rule.engine.transform.TbTransformMsgNode",
  22 + "name": "formResponse",
  23 + "debugMode": true,
  24 + "configuration": {
  25 + "jsScript": "if (msg.method == \"getResponse\") {\n return {msg: {\"response\": \"requestReceived\"}, metadata: metadata, msgType: msgType};\n}\n\nreturn {msg: msg, metadata: metadata, msgType: msgType};"
  26 + }
  27 + },
  28 + {
  29 + "additionalInfo": {
  30 + "layoutX": 450,
  31 + "layoutY": 300
  32 + },
  33 + "type": "org.thingsboard.rule.engine.rpc.TbSendRPCReplyNode",
  34 + "name": "rpcReply",
  35 + "debugMode": true,
  36 + "configuration": {
  37 + "requestIdMetaDataAttribute": "requestId"
  38 + }
  39 + }
  40 + ],
  41 + "connections": [
  42 + {
  43 + "fromIndex": 0,
  44 + "toIndex": 1,
  45 + "type": "RPC Request from Device"
  46 + },
  47 + {
  48 + "fromIndex": 1,
  49 + "toIndex": 2,
  50 + "type": "Success"
  51 + },
  52 + {
  53 + "fromIndex": 1,
  54 + "toIndex": 2,
  55 + "type": "Failure"
  56 + }
  57 + ],
  58 + "ruleChainConnections": null
  59 +}
\ No newline at end of file
... ...