From 5ffd778df110f0ec36f09f60348592c140d635e3 Mon Sep 17 00:00:00 2001 From: bombay Date: Sun, 7 Oct 2018 17:38:44 +0700 Subject: [PATCH 01/11] [WIP]Help refactor code solid patterns --- .../client/tools/MqttSslClient.java | 106 +++++++++++++----- 1 file changed, 77 insertions(+), 29 deletions(-) diff --git a/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java b/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java index dd27e44a1d..339f26ded8 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java +++ b/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java @@ -50,38 +50,86 @@ public class MqttSslClient { File ksFile = new File(ksUrl.toURI()); URL tsUrl = Resources.getResource(KEY_STORE_FILE); File tsFile = new File(tsUrl.toURI()); + char[] ksPwd = setKeyPassword(); + + KeyStore trustStore = setKeyStore(tsFile, ksPwd); + + TrustManagerFactory tmf = setTrustManagerFactory(trustStore); + + KeyStore ks = setKeyStore2(ksFile, ksPwd); + + KeyManagerFactory kmf = setKeyManagerFactory(ks); + + SSLContext sslContext = setSSLContext(kmf, tmf); + + MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(sslContext); + + MqttAsyncClient mqttAsyncClient = setMqttAsyncClient(mqttConnectOptions); + + setMessage(mqttAsyncClient, "v1/devices/me/telemetry", "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4}"); - TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); - - KeyStore trustStore = KeyStore.getInstance(JKS); - char[] ksPwd = new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x73, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; - trustStore.load(new FileInputStream(tsFile), ksPwd); - tmf.init(trustStore); - KeyStore ks = KeyStore.getInstance(JKS); - - ks.load(new FileInputStream(ksFile), ksPwd); - KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); - char[] clientPwd = new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x65, 0x79, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; - kmf.init(ks, clientPwd); - - KeyManager[] km = kmf.getKeyManagers(); - TrustManager[] tm = tmf.getTrustManagers(); - SSLContext sslContext = SSLContext.getInstance(TLS); - sslContext.init(km, tm, null); - - MqttConnectOptions options = new MqttConnectOptions(); - options.setSocketFactory(sslContext.getSocketFactory()); - MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, CLIENT_ID); - client.connect(options); - Thread.sleep(3000); - MqttMessage message = new MqttMessage(); - message.setPayload("{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4}".getBytes()); - client.publish("v1/devices/me/telemetry", message); - client.disconnect(); - log.info("Disconnected"); - System.exit(0); } catch (Exception e) { log.error("Unexpected exception occurred in MqttSslClient", e); } } + + private static char[] setKeyPassword(){ + return new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x73, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; + } + + private static KeyStore setKeyStore(File tsFile, char[] ksPwd) throws Exception{ + KeyStore trustStore = KeyStore.getInstance(JKS); + trustStore.load(new FileInputStream(tsFile), ksPwd); + return trustStore; + } + + private static TrustManagerFactory setTrustManagerFactory(KeyStore trustStore) throws Exception { + TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(trustStore); + return tmf; + } + + private static KeyStore setKeyStore2(File ksFile, char[] ksPwd)throws Exception{ + KeyStore ks = KeyStore.getInstance(JKS); + ks.load(new FileInputStream(ksFile), ksPwd); + return ks; + } + + private static KeyManagerFactory setKeyManagerFactory(KeyStore ks) throws Exception { + KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + char[] clientPwd = new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x65, 0x79, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; + kmf.init(ks, clientPwd); + return kmf; + } + + private static SSLContext setSSLContext(KeyManagerFactory kmf, TrustManagerFactory tmf) throws Exception { + KeyManager[] km = kmf.getKeyManagers(); + TrustManager[] tm = tmf.getTrustManagers(); + SSLContext sslContext = SSLContext.getInstance(TLS); + sslContext.init(km, tm, null); + return sslContext; + } + + private static MqttConnectOptions setMqttConnectOptions(SSLContext sslContext){ + MqttConnectOptions options = new MqttConnectOptions(); + options.setSocketFactory(sslContext.getSocketFactory()); + return options; + } + + private static MqttAsyncClient setMqttAsyncClient(MqttConnectOptions options) throws Exception{ + MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, CLIENT_ID); + client.connect(options); + Thread.sleep(3000); + return client; + } + + private static void setMessage(MqttAsyncClient client, String topic, String payload) throws Exception { + MqttMessage message = new MqttMessage(); + message.setPayload(payload.getBytes()); + client.publish(topic, message); + client.disconnect(); + log.info("Disconnected"); + System.exit(0); + } + } \ No newline at end of file From 4b0554b274714657f70a03938b36e4f9551fce8a Mon Sep 17 00:00:00 2001 From: Viacheslav Kukhtyn Date: Tue, 16 Oct 2018 14:05:52 +0300 Subject: [PATCH 02/11] Integration-tests subproject --- msa/integration-tests/README.md | 18 +++ msa/integration-tests/pom.xml | 98 +++++++++++++++ .../server/msa/AbstractContainerTest.java | 112 ++++++++++++++++++ .../server/msa/ContainerTestSuite.java | 37 ++++++ .../org/thingsboard/server/msa/WsClient.java | 57 +++++++++ .../server/msa/WsTelemetryResponse.java | 40 +++++++ .../msa/connectivity/HttpClientTest.java | 57 +++++++++ .../msa/connectivity/MqttClientTest.java | 111 +++++++++++++++++ msa/pom.xml | 1 + 9 files changed, 531 insertions(+) create mode 100644 msa/integration-tests/README.md create mode 100644 msa/integration-tests/pom.xml create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java diff --git a/msa/integration-tests/README.md b/msa/integration-tests/README.md new file mode 100644 index 0000000000..5ae6354740 --- /dev/null +++ b/msa/integration-tests/README.md @@ -0,0 +1,18 @@ + +## Integration tests execution +To run the integration tests with using Docker, the local Docker images of Thingsboard's microservices should be built.
+- Build the local Docker images in the directory with the Thingsboard's main [pom.xml](./../../pom.xml): + + mvn clean install -Ddockerfile.skip=false +- Verify that the new local images were built: + + docker image ls +As result, in REPOSITORY column, next images should be present: + + local-maven-build/tb-node + local-maven-build/tb-web-ui + local-maven-build/tb-web-ui + +- Run the integration tests in the [msa/integration-tests](../integration-tests) directory: + + mvn clean install -Dintegrationtests.skip=false \ No newline at end of file diff --git a/msa/integration-tests/pom.xml b/msa/integration-tests/pom.xml new file mode 100644 index 0000000000..a1c24f39c6 --- /dev/null +++ b/msa/integration-tests/pom.xml @@ -0,0 +1,98 @@ + + + 4.0.0 + + + org.thingsboard + 2.2.0-SNAPSHOT + msa + + org.thingsboard.msa + integration-tests + + ThingsBoard Integration Tests + https://thingsboard.io + Project for ThingsBoard integration tests with using Docker + + + UTF-8 + ${basedir}/../.. + true + 1.9.1 + 1.3.9 + + + + + org.testcontainers + testcontainers + ${testcontainers.version} + + + org.java-websocket + Java-WebSocket + ${java-websocket.version} + + + io.takari.junit + takari-cpsuite + + + ch.qos.logback + logback-classic + + + com.google.code.gson + gson + + + org.apache.commons + commons-lang3 + + + com.google.guava + guava + + + org.thingsboard + netty-mqtt + + + org.thingsboard + tools + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + **/*TestSuite.java + + ${integrationtests.skip} + + + + + + diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java new file mode 100644 index 0000000000..ab7f101f98 --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java @@ -0,0 +1,112 @@ +/** + * Copyright © 2016-2018 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.msa; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableMap; +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.RandomStringUtils; +import org.junit.*; +import org.thingsboard.client.tools.RestClient; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.id.DeviceId; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +@Slf4j +public abstract class AbstractContainerTest { + protected static String httpUrl; + protected static String wsUrl; + protected static RestClient restClient; + protected ObjectMapper mapper = new ObjectMapper(); + + @BeforeClass + public static void before() { + httpUrl = "http://localhost:" + ContainerTestSuite.composeContainer.getServicePort("tb-web-ui1", ContainerTestSuite.EXPOSED_PORT); + wsUrl = "ws://localhost:" + ContainerTestSuite.composeContainer.getServicePort("tb-web-ui1", ContainerTestSuite.EXPOSED_PORT); + restClient = new RestClient(httpUrl); + } + + protected Device createDevice(String name) { + return restClient.createDevice(name + RandomStringUtils.randomAlphanumeric(7), "DEFAULT"); + } + + protected WsClient subscribeToTelemetryWebSocket(DeviceId deviceId) throws URISyntaxException, InterruptedException { + WsClient mWs = new WsClient(new URI(wsUrl + "/api/ws/plugins/telemetry?token=" + restClient.getToken())); + mWs.connectBlocking(1, TimeUnit.SECONDS); + + JsonObject tsSubCmd = new JsonObject(); + tsSubCmd.addProperty("entityType", EntityType.DEVICE.name()); + tsSubCmd.addProperty("entityId", deviceId.toString()); + tsSubCmd.addProperty("scope", "LATEST_TELEMETRY"); + tsSubCmd.addProperty("cmdId", new Random().nextInt(100)); + tsSubCmd.addProperty("unsubscribe", false); + JsonArray wsTsSubCmds = new JsonArray(); + wsTsSubCmds.add(tsSubCmd); + JsonObject wsRequest = new JsonObject(); + wsRequest.add("tsSubCmds", wsTsSubCmds); + wsRequest.add("historyCmds", new JsonArray()); + wsRequest.add("attrSubCmds", new JsonArray()); + mWs.send(wsRequest.toString()); + return mWs; + } + + protected Map getExpectedLatestValues(long ts) { + return ImmutableMap.builder() + .put("booleanKey", ts) + .put("stringKey", ts) + .put("doubleKey", ts) + .put("longKey", ts) + .build(); + } + + protected boolean verify(WsTelemetryResponse wsTelemetryResponse, String key, Long expectedTs, String expectedValue) { + List list = wsTelemetryResponse.getDataValuesByKey(key); + return expectedTs.equals(list.get(0)) && expectedValue.equals(list.get(1)); + } + + protected boolean verify(WsTelemetryResponse wsTelemetryResponse, String key, String expectedValue) { + List list = wsTelemetryResponse.getDataValuesByKey(key); + return expectedValue.equals(list.get(1)); + } + + protected JsonObject createPayload(long ts) { + JsonObject values = createPayload(); + JsonObject payload = new JsonObject(); + payload.addProperty("ts", ts); + payload.add("values", values); + return payload; + } + + protected JsonObject createPayload() { + JsonObject values = new JsonObject(); + values.addProperty("stringKey", "value1"); + values.addProperty("booleanKey", true); + values.addProperty("doubleKey", 42.0); + values.addProperty("longKey", 73L); + + return values; + } + +} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java new file mode 100644 index 0000000000..fd2de229d0 --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2018 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.msa; + +import org.junit.ClassRule; +import org.junit.extensions.cpsuite.ClasspathSuite; +import org.junit.runner.RunWith; +import org.testcontainers.containers.DockerComposeContainer; +import org.testcontainers.containers.wait.strategy.Wait; + +import java.io.File; + +@RunWith(ClasspathSuite.class) +@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*"}) +public class ContainerTestSuite { + static final int EXPOSED_PORT = 8080; + + @ClassRule + public static DockerComposeContainer composeContainer = new DockerComposeContainer(new File("./../docker/docker-compose.yml")) + .withPull(false) + .withLocalCompose(true) + .withTailChildContainers(true) + .withExposedService("tb-web-ui1", EXPOSED_PORT, Wait.forHttp("/login")); +} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java new file mode 100644 index 0000000000..2f05eee4ce --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java @@ -0,0 +1,57 @@ +/** + * Copyright © 2016-2018 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.msa; + +import org.java_websocket.client.WebSocketClient; +import org.java_websocket.handshake.ServerHandshake; + +import java.net.URI; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +public class WsClient extends WebSocketClient { + private final BlockingQueue events; + private String message; + + public WsClient(URI serverUri) { + super(serverUri); + events = new ArrayBlockingQueue<>(100); + } + + @Override + public void onOpen(ServerHandshake serverHandshake) { + } + + @Override + public void onMessage(String message) { + events.add(message); + this.message = message; + } + + @Override + public void onClose(int code, String reason, boolean remote) { + events.clear(); + } + + @Override + public void onError(Exception ex) { + ex.printStackTrace(); + } + + public String getLastMessage() { + return this.message; + } +} \ No newline at end of file diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java new file mode 100644 index 0000000000..834c5d1449 --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java @@ -0,0 +1,40 @@ +/** + * Copyright © 2016-2018 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.msa; + +import lombok.Data; + +import java.io.Serializable; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +@Data +public class WsTelemetryResponse implements Serializable { + private int subscriptionId; + private int errorCode; + private String errorMsg; + private Map>> data; + private Map latestValues; + + public List getDataValuesByKey(String key) { + return data.entrySet().stream() + .filter(e -> e.getKey().equals(key)) + .flatMap(e -> e.getValue().stream().flatMap(Collection::stream)) + .collect(Collectors.toList()); + } +} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java new file mode 100644 index 0000000000..7cc0a0f6e3 --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java @@ -0,0 +1,57 @@ +/** + * Copyright © 2016-2018 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.msa.connectivity; + +import org.junit.Assert; +import org.junit.Test; +import org.springframework.http.ResponseEntity; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.security.DeviceCredentials; +import org.thingsboard.server.msa.AbstractContainerTest; +import org.thingsboard.server.msa.WsClient; +import org.thingsboard.server.msa.WsTelemetryResponse; + +import java.util.concurrent.TimeUnit; + +public class HttpClientTest extends AbstractContainerTest { + + @Test + public void telemetryUpdate() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + + Device device = createDevice("http_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + WsClient mWs = subscribeToTelemetryWebSocket(device.getId()); + ResponseEntity deviceTelemetryResponse = restClient.getRestTemplate() + .postForEntity(httpUrl + "/api/v1/{credentialsId}/telemetry", + mapper.readTree(createPayload().toString()), + ResponseEntity.class, + deviceCredentials.getCredentialsId()); + Assert.assertTrue(deviceTelemetryResponse.getStatusCode().is2xxSuccessful()); + TimeUnit.SECONDS.sleep(1); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class); + + Assert.assertEquals(getExpectedLatestValues(123456789L).keySet(), actualLatestTelemetry.getLatestValues().keySet()); + + Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString())); + Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1")); + Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0))); + Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73))); + + restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId()); + } +} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java new file mode 100644 index 0000000000..4ad638ed15 --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java @@ -0,0 +1,111 @@ +/** + * Copyright © 2016-2018 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.msa.connectivity; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import lombok.Data; +import org.junit.*; +import org.thingsboard.mqtt.MqttClient; +import org.thingsboard.mqtt.MqttClientConfig; +import org.thingsboard.mqtt.MqttHandler; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.security.DeviceCredentials; +import org.thingsboard.server.msa.AbstractContainerTest; +import org.thingsboard.server.msa.WsClient; +import org.thingsboard.server.msa.WsTelemetryResponse; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.*; + +public class MqttClientTest extends AbstractContainerTest { + + @Test + public void telemetryUpload() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + WsClient mWs = subscribeToTelemetryWebSocket(device.getId()); + MqttClient mqttClient = getMqttClient(deviceCredentials); + mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload().toString().getBytes())); + TimeUnit.SECONDS.sleep(1); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class); + + Assert.assertEquals(getExpectedLatestValues(123456789L).keySet(), actualLatestTelemetry.getLatestValues().keySet()); + + Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString())); + Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1")); + Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0))); + Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73))); + + restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId()); + } + + @Test + public void telemetryUploadWithTs() throws Exception { + long ts = 1451649600512L; + + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + WsClient mWs = subscribeToTelemetryWebSocket(device.getId()); + MqttClient mqttClient = getMqttClient(deviceCredentials); + mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload(ts).toString().getBytes())); + TimeUnit.SECONDS.sleep(1); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class); + + Assert.assertEquals(getExpectedLatestValues(ts), actualLatestTelemetry.getLatestValues()); + + Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", ts, Boolean.TRUE.toString())); + Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", ts, "value1")); + Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", ts, Double.toString(42.0))); + Assert.assertTrue(verify(actualLatestTelemetry, "longKey", ts, Long.toString(73))); + + restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId()); + } + + private MqttClient getMqttClient(DeviceCredentials deviceCredentials) throws InterruptedException { + MqttMessageListener queue = new MqttMessageListener(); + MqttClientConfig clientConfig = new MqttClientConfig(); + clientConfig.setClientId("MQTT client from test"); + clientConfig.setUsername(deviceCredentials.getCredentialsId()); + MqttClient mqttClient = MqttClient.create(clientConfig, queue); + mqttClient.connect("localhost", 1883).sync(); + return mqttClient; + } + + @Data + private class MqttMessageListener implements MqttHandler { + private final BlockingQueue events; + + private MqttMessageListener() { + events = new ArrayBlockingQueue<>(100); + } + + @Override + public void onMessage(String topic, ByteBuf message) { + events.add(new MqttEvent(topic, message.toString(StandardCharsets.UTF_8))); + } + } + + @Data + private class MqttEvent { + private final String topic; + private final String message; + } +} diff --git a/msa/pom.xml b/msa/pom.xml index 21ddb4dafe..dd4c3653ba 100644 --- a/msa/pom.xml +++ b/msa/pom.xml @@ -40,6 +40,7 @@ js-executor web-ui tb-node + integration-tests From fb098e1bae920b89358596ee274f4fd960673952 Mon Sep 17 00:00:00 2001 From: Viacheslav Kukhtyn Date: Tue, 23 Oct 2018 16:41:43 +0300 Subject: [PATCH 03/11] Cover MQTT API with black box tests --- msa/integration-tests/README.md | 2 +- msa/integration-tests/pom.xml | 10 +- .../server/msa/AbstractContainerTest.java | 113 +++++-- .../server/msa/ContainerTestSuite.java | 3 +- .../org/thingsboard/server/msa/WsClient.java | 8 +- .../msa/connectivity/HttpClientTest.java | 14 +- .../msa/connectivity/MqttClientTest.java | 304 +++++++++++++++++- .../server/msa/mapper/AttributesResponse.java | 26 ++ .../msa/{ => mapper}/WsTelemetryResponse.java | 2 +- .../RpcResponseRuleChainMetadata.json | 59 ++++ 10 files changed, 484 insertions(+), 57 deletions(-) create mode 100644 msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java rename msa/integration-tests/src/test/java/org/thingsboard/server/msa/{ => mapper}/WsTelemetryResponse.java (96%) create mode 100644 msa/integration-tests/src/test/resources/RpcResponseRuleChainMetadata.json diff --git a/msa/integration-tests/README.md b/msa/integration-tests/README.md index 5ae6354740..93df2bf77f 100644 --- a/msa/integration-tests/README.md +++ b/msa/integration-tests/README.md @@ -15,4 +15,4 @@ As result, in REPOSITORY column, next images should be present: - Run the integration tests in the [msa/integration-tests](../integration-tests) directory: - mvn clean install -Dintegrationtests.skip=false \ No newline at end of file + mvn clean install -DintegrationTests.skip=false \ No newline at end of file diff --git a/msa/integration-tests/pom.xml b/msa/integration-tests/pom.xml index a1c24f39c6..a9c67ccf93 100644 --- a/msa/integration-tests/pom.xml +++ b/msa/integration-tests/pom.xml @@ -34,9 +34,10 @@ UTF-8 ${basedir}/../.. - true + true 1.9.1 1.3.9 + 4.5.6 @@ -50,6 +51,11 @@ Java-WebSocket ${java-websocket.version} + + org.apache.httpcomponents + httpclient + ${httpclient.version} + io.takari.junit takari-cpsuite @@ -89,7 +95,7 @@ **/*TestSuite.java - ${integrationtests.skip} + ${integrationTests.skip} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java index ab7f101f98..5ebab780be 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java @@ -21,55 +21,68 @@ import com.google.gson.JsonArray; import com.google.gson.JsonObject; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.RandomStringUtils; +import org.apache.http.config.Registry; +import org.apache.http.config.RegistryBuilder; +import org.apache.http.conn.socket.ConnectionSocketFactory; +import org.apache.http.conn.ssl.SSLConnectionSocketFactory; +import org.apache.http.conn.ssl.TrustStrategy; +import org.apache.http.conn.ssl.X509HostnameVerifier; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; +import org.apache.http.ssl.SSLContextBuilder; +import org.apache.http.ssl.SSLContexts; import org.junit.*; +import org.springframework.http.client.HttpComponentsClientHttpRequestFactory; import org.thingsboard.client.tools.RestClient; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.msa.mapper.WsTelemetryResponse; +import javax.net.ssl.*; import java.net.URI; -import java.net.URISyntaxException; +import java.security.cert.X509Certificate; import java.util.List; import java.util.Map; import java.util.Random; -import java.util.concurrent.TimeUnit; @Slf4j public abstract class AbstractContainerTest { - protected static String httpUrl; - protected static String wsUrl; + protected static final String HTTPS_URL = "https://localhost"; + protected static final String WSS_URL = "wss://localhost"; protected static RestClient restClient; protected ObjectMapper mapper = new ObjectMapper(); @BeforeClass - public static void before() { - httpUrl = "http://localhost:" + ContainerTestSuite.composeContainer.getServicePort("tb-web-ui1", ContainerTestSuite.EXPOSED_PORT); - wsUrl = "ws://localhost:" + ContainerTestSuite.composeContainer.getServicePort("tb-web-ui1", ContainerTestSuite.EXPOSED_PORT); - restClient = new RestClient(httpUrl); + public static void before() throws Exception { + restClient = new RestClient(HTTPS_URL); + restClient.getRestTemplate().setRequestFactory(getRequestFactoryForSelfSignedCert()); } protected Device createDevice(String name) { return restClient.createDevice(name + RandomStringUtils.randomAlphanumeric(7), "DEFAULT"); } - protected WsClient subscribeToTelemetryWebSocket(DeviceId deviceId) throws URISyntaxException, InterruptedException { - WsClient mWs = new WsClient(new URI(wsUrl + "/api/ws/plugins/telemetry?token=" + restClient.getToken())); - mWs.connectBlocking(1, TimeUnit.SECONDS); - - JsonObject tsSubCmd = new JsonObject(); - tsSubCmd.addProperty("entityType", EntityType.DEVICE.name()); - tsSubCmd.addProperty("entityId", deviceId.toString()); - tsSubCmd.addProperty("scope", "LATEST_TELEMETRY"); - tsSubCmd.addProperty("cmdId", new Random().nextInt(100)); - tsSubCmd.addProperty("unsubscribe", false); - JsonArray wsTsSubCmds = new JsonArray(); - wsTsSubCmds.add(tsSubCmd); + protected WsClient subscribeToWebSocket(DeviceId deviceId, String scope, CmdsType property) throws Exception { + WsClient wsClient = new WsClient(new URI(WSS_URL + "/api/ws/plugins/telemetry?token=" + restClient.getToken())); + SSLContextBuilder builder = SSLContexts.custom(); + builder.loadTrustMaterial(null, (TrustStrategy) (chain, authType) -> true); + wsClient.setSocket(builder.build().getSocketFactory().createSocket()); + wsClient.connectBlocking(); + + JsonObject cmdsObject = new JsonObject(); + cmdsObject.addProperty("entityType", EntityType.DEVICE.name()); + cmdsObject.addProperty("entityId", deviceId.toString()); + cmdsObject.addProperty("scope", scope); + cmdsObject.addProperty("cmdId", new Random().nextInt(100)); + + JsonArray cmd = new JsonArray(); + cmd.add(cmdsObject); JsonObject wsRequest = new JsonObject(); - wsRequest.add("tsSubCmds", wsTsSubCmds); - wsRequest.add("historyCmds", new JsonArray()); - wsRequest.add("attrSubCmds", new JsonArray()); - mWs.send(wsRequest.toString()); - return mWs; + wsRequest.add(property.toString(), cmd); + wsClient.send(wsRequest.toString()); + return wsClient; } protected Map getExpectedLatestValues(long ts) { @@ -109,4 +122,54 @@ public abstract class AbstractContainerTest { return values; } + protected enum CmdsType { + TS_SUB_CMDS("tsSubCmds"), + HISTORY_CMDS("historyCmds"), + ATTR_SUB_CMDS("attrSubCmds"); + + private final String text; + + CmdsType(final String text) { + this.text = text; + } + + @Override + public String toString() { + return text; + } + } + + private static HttpComponentsClientHttpRequestFactory getRequestFactoryForSelfSignedCert() throws Exception { + SSLContextBuilder builder = SSLContexts.custom(); + builder.loadTrustMaterial(null, (TrustStrategy) (chain, authType) -> true); + SSLContext sslContext = builder.build(); + SSLConnectionSocketFactory sslSelfSigned = new SSLConnectionSocketFactory(sslContext, new X509HostnameVerifier() { + @Override + public void verify(String host, SSLSocket ssl) { + } + + @Override + public void verify(String host, X509Certificate cert) { + } + + @Override + public void verify(String host, String[] cns, String[] subjectAlts) { + } + + @Override + public boolean verify(String s, SSLSession sslSession) { + return true; + } + }); + + Registry socketFactoryRegistry = RegistryBuilder + .create() + .register("https", sslSelfSigned) + .build(); + + PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(socketFactoryRegistry); + CloseableHttpClient httpClient = HttpClients.custom().setConnectionManager(cm).build(); + return new HttpComponentsClientHttpRequestFactory(httpClient); + } + } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java index fd2de229d0..0629960b24 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java @@ -26,12 +26,11 @@ import java.io.File; @RunWith(ClasspathSuite.class) @ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*"}) public class ContainerTestSuite { - static final int EXPOSED_PORT = 8080; @ClassRule public static DockerComposeContainer composeContainer = new DockerComposeContainer(new File("./../docker/docker-compose.yml")) .withPull(false) .withLocalCompose(true) .withTailChildContainers(true) - .withExposedService("tb-web-ui1", EXPOSED_PORT, Wait.forHttp("/login")); + .withExposedService("tb-web-ui1", 8080, Wait.forHttp("/login")); } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java index 2f05eee4ce..5ef238f8aa 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java @@ -19,16 +19,12 @@ import org.java_websocket.client.WebSocketClient; import org.java_websocket.handshake.ServerHandshake; import java.net.URI; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; public class WsClient extends WebSocketClient { - private final BlockingQueue events; private String message; public WsClient(URI serverUri) { super(serverUri); - events = new ArrayBlockingQueue<>(100); } @Override @@ -37,13 +33,11 @@ public class WsClient extends WebSocketClient { @Override public void onMessage(String message) { - events.add(message); this.message = message; } @Override public void onClose(int code, String reason, boolean remote) { - events.clear(); } @Override @@ -54,4 +48,4 @@ public class WsClient extends WebSocketClient { public String getLastMessage() { return this.message; } -} \ No newline at end of file +} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java index 7cc0a0f6e3..f8e1041252 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.msa.connectivity; +import com.google.common.collect.Sets; import org.junit.Assert; import org.junit.Test; import org.springframework.http.ResponseEntity; @@ -22,7 +23,7 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.msa.AbstractContainerTest; import org.thingsboard.server.msa.WsClient; -import org.thingsboard.server.msa.WsTelemetryResponse; +import org.thingsboard.server.msa.mapper.WsTelemetryResponse; import java.util.concurrent.TimeUnit; @@ -35,23 +36,24 @@ public class HttpClientTest extends AbstractContainerTest { Device device = createDevice("http_"); DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); - WsClient mWs = subscribeToTelemetryWebSocket(device.getId()); + WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); ResponseEntity deviceTelemetryResponse = restClient.getRestTemplate() - .postForEntity(httpUrl + "/api/v1/{credentialsId}/telemetry", + .postForEntity(HTTPS_URL + "/api/v1/{credentialsId}/telemetry", mapper.readTree(createPayload().toString()), ResponseEntity.class, deviceCredentials.getCredentialsId()); Assert.assertTrue(deviceTelemetryResponse.getStatusCode().is2xxSuccessful()); TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); - Assert.assertEquals(getExpectedLatestValues(123456789L).keySet(), actualLatestTelemetry.getLatestValues().keySet()); + Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"), + actualLatestTelemetry.getLatestValues().keySet()); Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString())); Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1")); Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0))); Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73))); - restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId()); + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); } } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java index 4ad638ed15..eae98b9f1e 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java @@ -15,20 +15,39 @@ */ package org.thingsboard.server.msa.connectivity; +import com.fasterxml.jackson.databind.JsonNode; +import com.google.common.collect.Sets; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.gson.JsonObject; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import lombok.Data; +import org.apache.commons.lang3.RandomStringUtils; import org.junit.*; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; import org.thingsboard.mqtt.MqttClient; import org.thingsboard.mqtt.MqttClientConfig; import org.thingsboard.mqtt.MqttHandler; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.page.TextPageData; +import org.thingsboard.server.common.data.rule.NodeConnectionInfo; +import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.common.data.rule.RuleChainMetaData; +import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.msa.AbstractContainerTest; import org.thingsboard.server.msa.WsClient; -import org.thingsboard.server.msa.WsTelemetryResponse; +import org.thingsboard.server.msa.mapper.AttributesResponse; +import org.thingsboard.server.msa.mapper.WsTelemetryResponse; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.*; import java.util.concurrent.*; public class MqttClientTest extends AbstractContainerTest { @@ -39,20 +58,22 @@ public class MqttClientTest extends AbstractContainerTest { Device device = createDevice("mqtt_"); DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); - WsClient mWs = subscribeToTelemetryWebSocket(device.getId()); - MqttClient mqttClient = getMqttClient(deviceCredentials); + WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); + MqttClient mqttClient = getMqttClient(deviceCredentials, null); mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload().toString().getBytes())); TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); - Assert.assertEquals(getExpectedLatestValues(123456789L).keySet(), actualLatestTelemetry.getLatestValues().keySet()); + Assert.assertEquals(4, actualLatestTelemetry.getData().size()); + Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"), + actualLatestTelemetry.getLatestValues().keySet()); Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", Boolean.TRUE.toString())); Assert.assertTrue(verify(actualLatestTelemetry, "stringKey", "value1")); Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", Double.toString(42.0))); Assert.assertTrue(verify(actualLatestTelemetry, "longKey", Long.toString(73))); - restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId()); + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); } @Test @@ -63,12 +84,13 @@ public class MqttClientTest extends AbstractContainerTest { Device device = createDevice("mqtt_"); DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); - WsClient mWs = subscribeToTelemetryWebSocket(device.getId()); - MqttClient mqttClient = getMqttClient(deviceCredentials); + WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); + MqttClient mqttClient = getMqttClient(deviceCredentials, null); mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload(ts).toString().getBytes())); TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(mWs.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); + Assert.assertEquals(4, actualLatestTelemetry.getData().size()); Assert.assertEquals(getExpectedLatestValues(ts), actualLatestTelemetry.getLatestValues()); Assert.assertTrue(verify(actualLatestTelemetry, "booleanKey", ts, Boolean.TRUE.toString())); @@ -76,15 +98,271 @@ public class MqttClientTest extends AbstractContainerTest { Assert.assertTrue(verify(actualLatestTelemetry, "doubleKey", ts, Double.toString(42.0))); Assert.assertTrue(verify(actualLatestTelemetry, "longKey", ts, Long.toString(73))); - restClient.getRestTemplate().delete(httpUrl + "/api/device/" + device.getId()); + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); } - private MqttClient getMqttClient(DeviceCredentials deviceCredentials) throws InterruptedException { - MqttMessageListener queue = new MqttMessageListener(); + @Test + public void publishAttributeUpdateToServer() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + WsClient wsClient = subscribeToWebSocket(device.getId(), "CLIENT_SCOPE", CmdsType.ATTR_SUB_CMDS); + MqttMessageListener listener = new MqttMessageListener(); + MqttClient mqttClient = getMqttClient(deviceCredentials, listener); + JsonObject clientAttributes = new JsonObject(); + clientAttributes.addProperty("attr1", "value1"); + clientAttributes.addProperty("attr2", true); + clientAttributes.addProperty("attr3", 42.0); + clientAttributes.addProperty("attr4", 73); + mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes())); + TimeUnit.SECONDS.sleep(1); + WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); + + Assert.assertEquals(4, actualLatestTelemetry.getData().size()); + Assert.assertEquals(Sets.newHashSet("attr1", "attr2", "attr3", "attr4"), + actualLatestTelemetry.getLatestValues().keySet()); + + Assert.assertTrue(verify(actualLatestTelemetry, "attr1", "value1")); + Assert.assertTrue(verify(actualLatestTelemetry, "attr2", Boolean.TRUE.toString())); + Assert.assertTrue(verify(actualLatestTelemetry, "attr3", Double.toString(42.0))); + Assert.assertTrue(verify(actualLatestTelemetry, "attr4", Long.toString(73))); + + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); + } + + @Test + public void requestAttributeValuesFromServer() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + MqttMessageListener listener = new MqttMessageListener(); + MqttClient mqttClient = getMqttClient(deviceCredentials, listener); + + // Add a new client attribute + JsonObject clientAttributes = new JsonObject(); + String clientAttributeValue = RandomStringUtils.randomAlphanumeric(8); + clientAttributes.addProperty("clientAttr", clientAttributeValue); + mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes())); + + // Add a new shared attribute + JsonObject sharedAttributes = new JsonObject(); + String sharedAttributeValue = RandomStringUtils.randomAlphanumeric(8); + sharedAttributes.addProperty("sharedAttr", sharedAttributeValue); + ResponseEntity sharedAttributesResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE", + mapper.readTree(sharedAttributes.toString()), ResponseEntity.class, + device.getId()); + Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful()); + + // Subscribe to attributes response + mqttClient.on("v1/devices/me/attributes/response/+", listener); + // Request attributes + JsonObject request = new JsonObject(); + request.addProperty("clientKeys", "clientAttr"); + request.addProperty("sharedKeys", "sharedAttr"); + mqttClient.publish("v1/devices/me/attributes/request/" + new Random().nextInt(100), Unpooled.wrappedBuffer(request.toString().getBytes())); + MqttEvent event = listener.getEvents().poll(10, TimeUnit.SECONDS); + AttributesResponse attributes = mapper.readValue(Objects.requireNonNull(event).getMessage(), AttributesResponse.class); + + Assert.assertEquals(1, attributes.getClient().size()); + Assert.assertEquals(clientAttributeValue, attributes.getClient().get("clientAttr")); + + Assert.assertEquals(1, attributes.getShared().size()); + Assert.assertEquals(sharedAttributeValue, attributes.getShared().get("sharedAttr")); + + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); + } + + @Test + public void subscribeToAttributeUpdatesFromServer() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + MqttMessageListener listener = new MqttMessageListener(); + MqttClient mqttClient = getMqttClient(deviceCredentials, listener); + mqttClient.on("v1/devices/me/attributes", listener); + + String sharedAttributeName = "sharedAttr"; + + // Add a new shared attribute + JsonObject sharedAttributes = new JsonObject(); + String sharedAttributeValue = RandomStringUtils.randomAlphanumeric(8); + sharedAttributes.addProperty(sharedAttributeName, sharedAttributeValue); + ResponseEntity sharedAttributesResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE", + mapper.readTree(sharedAttributes.toString()), ResponseEntity.class, + device.getId()); + Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful()); + + MqttEvent event = listener.getEvents().poll(10, TimeUnit.SECONDS); + Assert.assertEquals(sharedAttributeValue, + mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText()); + + // Update the shared attribute value + JsonObject updatedSharedAttributes = new JsonObject(); + String updatedSharedAttributeValue = RandomStringUtils.randomAlphanumeric(8); + updatedSharedAttributes.addProperty(sharedAttributeName, updatedSharedAttributeValue); + ResponseEntity updatedSharedAttributesResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE", + mapper.readTree(updatedSharedAttributes.toString()), ResponseEntity.class, + device.getId()); + Assert.assertTrue(updatedSharedAttributesResponse.getStatusCode().is2xxSuccessful()); + + event = listener.getEvents().poll(10, TimeUnit.SECONDS); + Assert.assertEquals(updatedSharedAttributeValue, + mapper.readValue(Objects.requireNonNull(event).getMessage(), JsonNode.class).get(sharedAttributeName).asText()); + + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); + } + + @Test + public void serverSideRpc() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + MqttMessageListener listener = new MqttMessageListener(); + MqttClient mqttClient = getMqttClient(deviceCredentials, listener); + mqttClient.on("v1/devices/me/rpc/request/+", listener); + + // Send an RPC from the server + JsonObject serverRpcPayload = new JsonObject(); + serverRpcPayload.addProperty("method", "getValue"); + serverRpcPayload.addProperty("params", true); + serverRpcPayload.addProperty("timeout", 1000); + ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); + ListenableFuture future = service.submit(() -> { + try { + return restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/plugins/rpc/twoway/{deviceId}", + mapper.readTree(serverRpcPayload.toString()), String.class, + device.getId()); + } catch (IOException e) { + return ResponseEntity.badRequest().build(); + } + }); + + // Wait for RPC call from the server and send the response + MqttEvent requestFromServer = listener.getEvents().poll(10, TimeUnit.SECONDS); + + Assert.assertEquals("{\"method\":\"getValue\",\"params\":true}", Objects.requireNonNull(requestFromServer).getMessage()); + + Integer requestId = Integer.valueOf(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length())); + JsonObject clientResponse = new JsonObject(); + clientResponse.addProperty("response", "someResponse"); + // Send a response to the server's RPC request + mqttClient.publish("v1/devices/me/rpc/response/" + requestId, Unpooled.wrappedBuffer(clientResponse.toString().getBytes())); + + ResponseEntity serverResponse = future.get(5, TimeUnit.SECONDS); + Assert.assertTrue(serverResponse.getStatusCode().is2xxSuccessful()); + Assert.assertEquals(clientResponse.toString(), serverResponse.getBody()); + + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); + } + + @Test + public void clientSideRpc() throws Exception { + restClient.login("tenant@thingsboard.org", "tenant"); + Device device = createDevice("mqtt_"); + DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); + + MqttMessageListener listener = new MqttMessageListener(); + MqttClient mqttClient = getMqttClient(deviceCredentials, listener); + mqttClient.on("v1/devices/me/rpc/request/+", listener); + + // Get the default rule chain id to make it root again after test finished + RuleChainId defaultRuleChainId = getDefaultRuleChainId(); + + // Create a new root rule chain + RuleChainId ruleChainId = createRootRuleChainForRpcResponse(); + + // Send the request to the server + JsonObject clientRequest = new JsonObject(); + clientRequest.addProperty("method", "getResponse"); + clientRequest.addProperty("params", true); + Integer requestId = 42; + mqttClient.publish("v1/devices/me/rpc/request/" + requestId, Unpooled.wrappedBuffer(clientRequest.toString().getBytes())); + + // Check the response from the server + TimeUnit.SECONDS.sleep(1); + MqttEvent responseFromServer = listener.getEvents().poll(1, TimeUnit.SECONDS); + Integer responseId = Integer.valueOf(Objects.requireNonNull(responseFromServer).getTopic().substring("v1/devices/me/rpc/response/".length())); + Assert.assertEquals(requestId, responseId); + Assert.assertEquals("requestReceived", mapper.readTree(responseFromServer.getMessage()).get("response").asText()); + + // Make the default rule chain a root again + ResponseEntity rootRuleChainResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/ruleChain/{ruleChainId}/root", + null, + RuleChain.class, + defaultRuleChainId); + Assert.assertTrue(rootRuleChainResponse.getStatusCode().is2xxSuccessful()); + + // Delete the created rule chain + restClient.getRestTemplate().delete(HTTPS_URL + "/api/ruleChain/{ruleChainId}", ruleChainId); + restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId()); + } + + private RuleChainId createRootRuleChainForRpcResponse() throws Exception { + RuleChain newRuleChain = new RuleChain(); + newRuleChain.setName("testRuleChain"); + ResponseEntity ruleChainResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/ruleChain", + newRuleChain, + RuleChain.class); + Assert.assertTrue(ruleChainResponse.getStatusCode().is2xxSuccessful()); + RuleChain ruleChain = ruleChainResponse.getBody(); + + JsonNode configuration = mapper.readTree(this.getClass().getClassLoader().getResourceAsStream("RpcResponseRuleChainMetadata.json")); + RuleChainMetaData ruleChainMetaData = new RuleChainMetaData(); + ruleChainMetaData.setRuleChainId(ruleChain.getId()); + ruleChainMetaData.setFirstNodeIndex(configuration.get("firstNodeIndex").asInt()); + ruleChainMetaData.setNodes(Arrays.asList(mapper.treeToValue(configuration.get("nodes"), RuleNode[].class))); + ruleChainMetaData.setConnections(Arrays.asList(mapper.treeToValue(configuration.get("connections"), NodeConnectionInfo[].class))); + + ResponseEntity ruleChainMetadataResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/ruleChain/metadata", + ruleChainMetaData, + RuleChainMetaData.class); + Assert.assertTrue(ruleChainMetadataResponse.getStatusCode().is2xxSuccessful()); + + // Set a new rule chain as root + ResponseEntity rootRuleChainResponse = restClient.getRestTemplate() + .postForEntity(HTTPS_URL + "/api/ruleChain/{ruleChainId}/root", + null, + RuleChain.class, + ruleChain.getId()); + Assert.assertTrue(rootRuleChainResponse.getStatusCode().is2xxSuccessful()); + + return ruleChain.getId(); + } + + private RuleChainId getDefaultRuleChainId() { + ResponseEntity> ruleChains = restClient.getRestTemplate().exchange( + HTTPS_URL + "/api/ruleChains?limit=40&textSearch=", + HttpMethod.GET, + null, + new ParameterizedTypeReference>() { + }); + + Optional defaultRuleChain = ruleChains.getBody().getData() + .stream() + .filter(RuleChain::isRoot) + .findFirst(); + if (!defaultRuleChain.isPresent()) { + Assert.fail("Root rule chain wasn't found"); + } + return defaultRuleChain.get().getId(); + } + + private MqttClient getMqttClient(DeviceCredentials deviceCredentials, MqttMessageListener listener) throws InterruptedException { MqttClientConfig clientConfig = new MqttClientConfig(); clientConfig.setClientId("MQTT client from test"); clientConfig.setUsername(deviceCredentials.getCredentialsId()); - MqttClient mqttClient = MqttClient.create(clientConfig, queue); + MqttClient mqttClient = MqttClient.create(clientConfig, listener); mqttClient.connect("localhost", 1883).sync(); return mqttClient; } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java new file mode 100644 index 0000000000..f9774eef20 --- /dev/null +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2018 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.msa.mapper; + +import lombok.Data; + +import java.util.Map; + +@Data +public class AttributesResponse { + private Map client; + private Map shared; +} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java similarity index 96% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java rename to msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java index 834c5d1449..b22244f3a7 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsTelemetryResponse.java +++ b/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.msa; +package org.thingsboard.server.msa.mapper; import lombok.Data; diff --git a/msa/integration-tests/src/test/resources/RpcResponseRuleChainMetadata.json b/msa/integration-tests/src/test/resources/RpcResponseRuleChainMetadata.json new file mode 100644 index 0000000000..09178ef781 --- /dev/null +++ b/msa/integration-tests/src/test/resources/RpcResponseRuleChainMetadata.json @@ -0,0 +1,59 @@ +{ + "firstNodeIndex": 0, + "nodes": [ + { + "additionalInfo": { + "layoutX": 325, + "layoutY": 150 + }, + "type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode", + "name": "msgTypeSwitch", + "debugMode": true, + "configuration": { + "version": 0 + } + }, + { + "additionalInfo": { + "layoutX": 60, + "layoutY": 300 + }, + "type": "org.thingsboard.rule.engine.transform.TbTransformMsgNode", + "name": "formResponse", + "debugMode": true, + "configuration": { + "jsScript": "if (msg.method == \"getResponse\") {\n return {msg: {\"response\": \"requestReceived\"}, metadata: metadata, msgType: msgType};\n}\n\nreturn {msg: msg, metadata: metadata, msgType: msgType};" + } + }, + { + "additionalInfo": { + "layoutX": 450, + "layoutY": 300 + }, + "type": "org.thingsboard.rule.engine.rpc.TbSendRPCReplyNode", + "name": "rpcReply", + "debugMode": true, + "configuration": { + "requestIdMetaDataAttribute": "requestId" + } + } + ], + "connections": [ + { + "fromIndex": 0, + "toIndex": 1, + "type": "RPC Request from Device" + }, + { + "fromIndex": 1, + "toIndex": 2, + "type": "Success" + }, + { + "fromIndex": 1, + "toIndex": 2, + "type": "Failure" + } + ], + "ruleChainConnections": null +} \ No newline at end of file From 7d14d46edf693a25a933cc57fe1829bbce090730 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yusuf=20M=C3=BCcahit=20=C3=87etinkaya?= Date: Fri, 26 Oct 2018 19:31:35 +0300 Subject: [PATCH 04/11] check and recover currentServer zNode on zk. --- .../cluster/discovery/ZkDiscoveryService.java | 48 +++++++++++++++---- 1 file changed, 39 insertions(+), 9 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java index 6f51ceed26..c1876d6b1b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java @@ -29,6 +29,7 @@ import org.apache.curator.framework.state.ConnectionStateListener; import org.apache.curator.retry.RetryForever; import org.apache.curator.utils.CloseableUtils; import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -49,6 +50,8 @@ import java.util.NoSuchElementException; import java.util.concurrent.CopyOnWriteArrayList; import java.util.stream.Collectors; +import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.CHILD_REMOVED; + /** * @author Andrew Shvayka */ @@ -121,19 +124,42 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi } @Override - public void publishCurrentServer() { + public synchronized void publishCurrentServer() { + ServerInstance self = this.serverInstance.getSelf(); + if (currentServerExists()) { + log.info("[{}:{}] ZK node for current instance already exists, NOT created new one: {}", self.getHost(), self.getPort(), nodePath); + } else { + try { + log.info("[{}:{}] Creating ZK node for current instance", self.getHost(), self.getPort()); + nodePath = client.create() + .creatingParentsIfNeeded() + .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zkNodesDir + "/", SerializationUtils.serialize(self.getServerAddress())); + log.info("[{}:{}] Created ZK node for current instance: {}", self.getHost(), self.getPort(), nodePath); + client.getConnectionStateListenable().addListener(checkReconnect(self)); + } catch (Exception e) { + log.error("Failed to create ZK node", e); + throw new RuntimeException(e); + } + } + } + + private boolean currentServerExists() { + if (nodePath == null) { + return false; + } try { ServerInstance self = this.serverInstance.getSelf(); - log.info("[{}:{}] Creating ZK node for current instance", self.getHost(), self.getPort()); - nodePath = client.create() - .creatingParentsIfNeeded() - .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zkNodesDir + "/", SerializationUtils.serialize(self.getServerAddress())); - log.info("[{}:{}] Created ZK node for current instance: {}", self.getHost(), self.getPort(), nodePath); - client.getConnectionStateListenable().addListener(checkReconnect(self)); + ServerAddress registeredServerAdress = null; + registeredServerAdress = SerializationUtils.deserialize(client.getData().forPath(nodePath)); + if (self.getServerAddress() != null && self.getServerAddress().equals(registeredServerAdress)) { + return true; + } + } catch (KeeperException.NoNodeException e) { + log.info("ZK node does not exist: {}", nodePath); } catch (Exception e) { - log.error("Failed to create ZK node", e); - throw new RuntimeException(e); + log.error("Couldn't check if ZK node exists", e); } + return false; } private ConnectionStateListener checkReconnect(ServerInstance self) { @@ -221,6 +247,10 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi log.debug("Ignoring {} due to empty child's data", pathChildrenCacheEvent); return; } else if (nodePath != null && nodePath.equals(data.getPath())) { + if (pathChildrenCacheEvent.getType() == CHILD_REMOVED) { + log.info("ZK node for current instance is somehow deleted."); + publishCurrentServer(); + } log.debug("Ignoring event about current server {}", pathChildrenCacheEvent); return; } From 093eafb462cab211107761c6fa54f1aba54835cb Mon Sep 17 00:00:00 2001 From: Dima Landiak Date: Mon, 29 Oct 2018 17:48:20 +0200 Subject: [PATCH 05/11] delete timeseries fixes and controller, entityView findByName added --- .../controller/EntityViewController.java | 20 +++-- .../controller/TelemetryController.java | 67 +++++++++++++++- .../server/common/data/audit/ActionType.java | 11 +-- .../dao/entityview/EntityViewService.java | 2 + .../dao/entityview/EntityViewServiceImpl.java | 19 +++-- .../dao/sql/timeseries/JpaTimeseriesDao.java | 79 +++++++++++++++++-- .../dao/sql/timeseries/TsKvRepository.java | 2 +- .../CassandraBaseTimeseriesDao.java | 46 +++++++---- .../timeseries/BaseTimeseriesServiceTest.java | 22 +++++- 9 files changed, 224 insertions(+), 44 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java index 9b7ffd2d11..2a9c073d34 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java @@ -29,7 +29,6 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; -import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntitySubtype; @@ -39,7 +38,6 @@ import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.entityview.EntityViewSearchQuery; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.CustomerId; -import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; @@ -47,7 +45,6 @@ import org.thingsboard.server.common.data.id.UUIDBased; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.page.TextPageData; import org.thingsboard.server.common.data.page.TextPageLink; -import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; import org.thingsboard.server.dao.exception.IncorrectParameterException; import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.service.security.model.SecurityUser; @@ -174,7 +171,7 @@ public class EntityViewController extends BaseController { EntityView entityView = checkEntityViewId(entityViewId); entityViewService.deleteEntityView(entityViewId); logEntityAction(entityViewId, entityView, entityView.getCustomerId(), - ActionType.DELETED,null, strEntityViewId); + ActionType.DELETED, null, strEntityViewId); } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, @@ -184,11 +181,24 @@ public class EntityViewController extends BaseController { } } + @PreAuthorize("hasAuthority('TENANT_ADMIN')") + @RequestMapping(value = "/tenant/entityViews", params = {"entityViewName"}, method = RequestMethod.GET) + @ResponseBody + public EntityView getTenantEntityView( + @RequestParam String entityViewName) throws ThingsboardException { + try { + TenantId tenantId = getCurrentUser().getTenantId(); + return checkNotNull(entityViewService.findEntityViewByTenantIdAndName(tenantId, entityViewName)); + } catch (Exception e) { + throw handleException(e); + } + } + @PreAuthorize("hasAuthority('TENANT_ADMIN')") @RequestMapping(value = "/customer/{customerId}/entityView/{entityViewId}", method = RequestMethod.POST) @ResponseBody public EntityView assignEntityViewToCustomer(@PathVariable(CUSTOMER_ID) String strCustomerId, - @PathVariable(ENTITY_VIEW_ID) String strEntityViewId) throws ThingsboardException { + @PathVariable(ENTITY_VIEW_ID) String strEntityViewId) throws ThingsboardException { checkParameter(CUSTOMER_ID, strCustomerId); checkParameter(ENTITY_VIEW_ID, strEntityViewId); try { diff --git a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java index 6c37614147..ef09a7a62c 100644 --- a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java @@ -49,9 +49,11 @@ import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.kv.AttributeKey; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseDeleteTsKvQuery; import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; +import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; import org.thingsboard.server.common.data.kv.DoubleDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; @@ -60,12 +62,10 @@ import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; import org.thingsboard.server.common.transport.adaptor.JsonConverter; -import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.service.security.AccessValidator; import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.telemetry.AttributeData; -import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import org.thingsboard.server.service.telemetry.TsData; import org.thingsboard.server.service.telemetry.exception.InvalidParametersException; import org.thingsboard.server.service.telemetry.exception.UncheckedApiException; @@ -249,6 +249,60 @@ public class TelemetryController extends BaseController { return saveTelemetry(entityId, requestBody, ttl); } + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{entityType}/{entityId}/timeseries/delete", method = RequestMethod.DELETE) + @ResponseBody + public DeferredResult deleteEntityTimeseries(@PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, + @RequestParam(name = "keys") String keysStr, + @RequestParam(name = "deleteAllDataForKeys", defaultValue = "false") boolean deleteAllDataForKeys, + @RequestParam(name = "startTs", required = false) Long startTs, + @RequestParam(name = "endTs", required = false) Long endTs, + @RequestParam(name = "rewriteLatestIfDeleted", defaultValue = "false") boolean rewriteLatestIfDeleted) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); + return deleteTimeseries(entityId, keysStr, deleteAllDataForKeys, startTs, endTs, rewriteLatestIfDeleted); + } + + private DeferredResult deleteTimeseries(EntityId entityIdStr, String keysStr, boolean deleteAllDataForKeys, + Long startTs, Long endTs, boolean rewriteLatestIfDeleted) throws ThingsboardException { + List keys = toKeysList(keysStr); + if (keys.isEmpty()) { + return getImmediateDeferredResult("Empty keys: " + keysStr, HttpStatus.BAD_REQUEST); + } + SecurityUser user = getCurrentUser(); + + long deleteFromTs; + long deleteToTs; + if (deleteAllDataForKeys) { + deleteFromTs = 0L; + deleteToTs = System.currentTimeMillis(); + } else { + deleteFromTs = startTs; + deleteToTs = endTs; + } + + return accessValidator.validateEntityAndCallback(user, entityIdStr, (result, entityId) -> { + List deleteTsKvQueries = new ArrayList<>(); + for (String key : keys) { + deleteTsKvQueries.add(new BaseDeleteTsKvQuery(key, deleteFromTs, deleteToTs, rewriteLatestIfDeleted)); + } + + ListenableFuture> future = tsService.remove(entityId, deleteTsKvQueries); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List tmp) { + logTimeseriesDeleted(user, entityId, keys, null); + result.setResult(new ResponseEntity<>(HttpStatus.OK)); + } + + @Override + public void onFailure(Throwable t) { + logTimeseriesDeleted(user, entityId, keys, t); + result.setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR)); + } + }, executor); + }); + } + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{deviceId}/{scope}", method = RequestMethod.DELETE) @ResponseBody @@ -506,6 +560,15 @@ public class TelemetryController extends BaseController { }; } + private void logTimeseriesDeleted(SecurityUser user, EntityId entityId, List keys, Throwable e) { + try { + logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, ActionType.TIMESERIES_DELETED, toException(e), + keys); + } catch (ThingsboardException te) { + log.warn("Failed to log timeseries delete", te); + } + } + private void logAttributesDeleted(SecurityUser user, EntityId entityId, String scope, List keys, Throwable e) { try { logEntityAction(user, (UUIDBased & EntityId) entityId, null, null, ActionType.ATTRIBUTES_DELETED, toException(e), diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/audit/ActionType.java b/common/data/src/main/java/org/thingsboard/server/common/data/audit/ActionType.java index c37d4607c3..822387cfcb 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/audit/ActionType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/audit/ActionType.java @@ -24,6 +24,7 @@ public enum ActionType { UPDATED(false), // log entity ATTRIBUTES_UPDATED(false), // log attributes/values ATTRIBUTES_DELETED(false), // log attributes + TIMESERIES_DELETED(false), // log timeseries RPC_CALL(false), // log method and params CREDENTIALS_UPDATED(false), // log new credentials ASSIGNED_TO_CUSTOMER(false), // log customer name @@ -32,11 +33,11 @@ public enum ActionType { SUSPENDED(false), // log string id CREDENTIALS_READ(true), // log device id ATTRIBUTES_READ(true), // log attributes - RELATION_ADD_OR_UPDATE (false), - RELATION_DELETED (false), - RELATIONS_DELETED (false), - ALARM_ACK (false), - ALARM_CLEAR (false); + RELATION_ADD_OR_UPDATE(false), + RELATION_DELETED(false), + RELATIONS_DELETED(false), + ALARM_ACK(false), + ALARM_CLEAR(false); private final boolean isRead; diff --git a/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewService.java b/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewService.java index da87b44322..9c828663c1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewService.java @@ -43,6 +43,8 @@ public interface EntityViewService { EntityView findEntityViewById(EntityViewId entityViewId); + EntityView findEntityViewByTenantIdAndName(TenantId tenantId, String name); + TextPageData findEntityViewByTenantId(TenantId tenantId, TextPageLink pageLink); TextPageData findEntityViewByTenantIdAndType(TenantId tenantId, TextPageLink pageLink, String type); diff --git a/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java index 2d94cc2eab..9f8949dc4e 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java @@ -29,8 +29,6 @@ import org.springframework.cache.annotation.Cacheable; import org.springframework.cache.annotation.Caching; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.Customer; -import org.thingsboard.server.common.data.DataConstants; -import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; @@ -40,12 +38,10 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.page.TextPageData; import org.thingsboard.server.common.data.page.TextPageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.EntitySearchDirection; -import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.customer.CustomerDao; import org.thingsboard.server.dao.entity.AbstractEntityService; import org.thingsboard.server.dao.exception.DataValidationException; @@ -56,15 +52,13 @@ import org.thingsboard.server.dao.tenant.TenantDao; import javax.annotation.Nullable; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.List; -import java.util.concurrent.ExecutionException; +import java.util.Optional; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.CacheConstants.ENTITY_VIEW_CACHE; -import static org.thingsboard.server.common.data.CacheConstants.RELATIONS_CACHE; import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; import static org.thingsboard.server.dao.service.Validator.validateId; import static org.thingsboard.server.dao.service.Validator.validatePageLink; @@ -96,6 +90,7 @@ public class EntityViewServiceImpl extends AbstractEntityService implements Enti @Caching(evict = { @CacheEvict(cacheNames = ENTITY_VIEW_CACHE, key = "{#entityView.tenantId, #entityView.entityId}"), + @CacheEvict(cacheNames = ENTITY_VIEW_CACHE, key = "{#entityView.tenantId, #entityView.name}"), @CacheEvict(cacheNames = ENTITY_VIEW_CACHE, key = "{#entityView.id}")}) @Override public EntityView saveEntityView(EntityView entityView) { @@ -137,6 +132,15 @@ public class EntityViewServiceImpl extends AbstractEntityService implements Enti return entityViewDao.findById(entityViewId.getId()); } + @Cacheable(cacheNames = ENTITY_VIEW_CACHE, key = "{#tenantId, #name}") + @Override + public EntityView findEntityViewByTenantIdAndName(TenantId tenantId, String name) { + log.trace("Executing findEntityViewByTenantIdAndName [{}][{}]", tenantId, name); + validateId(tenantId, INCORRECT_TENANT_ID + tenantId); + Optional entityViewOpt = entityViewDao.findEntityViewByTenantIdAndName(tenantId.getId(), name); + return entityViewOpt.orElse(null); + } + @Override public TextPageData findEntityViewByTenantId(TenantId tenantId, TextPageLink pageLink) { log.trace("Executing findEntityViewsByTenantId, tenantId [{}], pageLink [{}]", tenantId, pageLink); @@ -255,6 +259,7 @@ public class EntityViewServiceImpl extends AbstractEntityService implements Enti deleteEntityRelations(entityViewId); EntityView entityView = entityViewDao.findById(entityViewId.getId()); cacheManager.getCache(ENTITY_VIEW_CACHE).evict(Arrays.asList(entityView.getTenantId(), entityView.getEntityId())); + cacheManager.getCache(ENTITY_VIEW_CACHE).evict(Arrays.asList(entityView.getTenantId(), entityView.getName())); entityViewDao.removeById(entityViewId.getId()); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java index 5bd9175628..04227a3825 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java @@ -17,6 +17,7 @@ package org.thingsboard.server.dao.sql.timeseries; import com.google.common.base.Function; import com.google.common.collect.Lists; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; @@ -31,6 +32,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.UUIDConverter; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; import org.thingsboard.server.common.data.kv.ReadTsKvQuery; @@ -41,9 +43,9 @@ import org.thingsboard.server.dao.model.sql.TsKvEntity; import org.thingsboard.server.dao.model.sql.TsKvLatestCompositeKey; import org.thingsboard.server.dao.model.sql.TsKvLatestEntity; import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService; +import org.thingsboard.server.dao.timeseries.SimpleListenableFuture; import org.thingsboard.server.dao.timeseries.TimeseriesDao; import org.thingsboard.server.dao.timeseries.TsInsertExecutorType; -import org.thingsboard.server.dao.util.SqlDao; import org.thingsboard.server.dao.util.SqlTsDao; import javax.annotation.Nullable; @@ -53,6 +55,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.stream.Collectors; @@ -64,6 +67,8 @@ import static org.thingsboard.server.common.data.UUIDConverter.fromTimeUUID; @SqlTsDao public class JpaTimeseriesDao extends JpaAbstractDaoListeningExecutorService implements TimeseriesDao { + private static final String DESC_ORDER = "DESC"; + @Value("${sql.ts_inserts_executor_type}") private String insertExecutorType; @@ -326,14 +331,72 @@ public class JpaTimeseriesDao extends JpaAbstractDaoListeningExecutorService imp @Override public ListenableFuture removeLatest(EntityId entityId, DeleteTsKvQuery query) { - TsKvLatestEntity latestEntity = new TsKvLatestEntity(); - latestEntity.setEntityType(entityId.getEntityType()); - latestEntity.setEntityId(fromTimeUUID(entityId.getId())); - latestEntity.setKey(query.getKey()); - return service.submit(() -> { - tsKvLatestRepository.delete(latestEntity); - return null; + ListenableFuture latestFuture = findLatest(entityId, query.getKey()); + + ListenableFuture booleanFuture = Futures.transform(latestFuture, tsKvEntry -> { + long ts = tsKvEntry.getTs(); + return ts > query.getStartTs() && ts <= query.getEndTs(); + }, service); + + ListenableFuture removedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { + if (isRemove) { + TsKvLatestEntity latestEntity = new TsKvLatestEntity(); + latestEntity.setEntityType(entityId.getEntityType()); + latestEntity.setEntityId(fromTimeUUID(entityId.getId())); + latestEntity.setKey(query.getKey()); + return service.submit(() -> { + tsKvLatestRepository.delete(latestEntity); + return null; + }); + } + return Futures.immediateFuture(null); + }, service); + + final SimpleListenableFuture resultFuture = new SimpleListenableFuture<>(); + Futures.addCallback(removedLatestFuture, new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + if (query.getRewriteLatestIfDeleted()) { + ListenableFuture savedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { + if (isRemove) { + return getNewLatestEntryFuture(entityId, query); + } + return Futures.immediateFuture(null); + }, service); + + try { + resultFuture.set(savedLatestFuture.get()); + } catch (InterruptedException | ExecutionException e) { + log.warn("Could not get latest saved value for [{}], {}", entityId, query.getKey(), e); + } + } else { + resultFuture.set(null); + } + } + + @Override + public void onFailure(Throwable t) { + log.warn("[{}] Failed to process remove of the latest value", entityId, t); + } }); + return resultFuture; + } + + private ListenableFuture getNewLatestEntryFuture(EntityId entityId, DeleteTsKvQuery query) { + long startTs = 0; + long endTs = query.getStartTs() - 1; + ReadTsKvQuery findNewLatestQuery = new BaseReadTsKvQuery(query.getKey(), startTs, endTs, endTs - startTs, 1, + Aggregation.NONE, DESC_ORDER); + ListenableFuture> future = findAllAsync(entityId, findNewLatestQuery); + + return Futures.transformAsync(future, entryList -> { + if (entryList.size() == 1) { + return saveLatest(entityId, entryList.get(0)); + } else { + log.trace("Could not find new latest value for [{}], key - {}", entityId, query.getKey()); + } + return Futures.immediateFuture(null); + }, service); } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java index 4c743e5ab6..296d173cf7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java @@ -47,7 +47,7 @@ public interface TsKvRepository extends CrudRepository :startTs AND tskv.ts < :endTs") + "AND tskv.ts > :startTs AND tskv.ts <= :endTs") void delete(@Param("entityId") String entityId, @Param("entityType") EntityType entityType, @Param("entityKey") String key, diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java index 709bfd5432..fdc69f9262 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java @@ -48,7 +48,6 @@ import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.dao.nosql.CassandraAbstractAsyncDao; -import org.thingsboard.server.dao.util.NoSqlDao; import org.thingsboard.server.dao.util.NoSqlTsDao; import javax.annotation.Nullable; @@ -62,6 +61,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Optional; +import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import static com.datastax.driver.core.querybuilder.QueryBuilder.eq; @@ -434,14 +434,14 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem public ListenableFuture removeLatest(EntityId entityId, DeleteTsKvQuery query) { ListenableFuture latestEntryFuture = findLatest(entityId, query.getKey()); - ListenableFuture booleanFuture = Futures.transformAsync(latestEntryFuture, latestEntry -> { + ListenableFuture booleanFuture = Futures.transform(latestEntryFuture, latestEntry -> { long ts = latestEntry.getTs(); - if (ts >= query.getStartTs() && ts <= query.getEndTs()) { - return Futures.immediateFuture(true); + if (ts > query.getStartTs() && ts <= query.getEndTs()) { + return true; } else { log.trace("Won't be deleted latest value for [{}], key - {}", entityId, query.getKey()); } - return Futures.immediateFuture(false); + return false; }, readResultsProcessingExecutor); ListenableFuture removedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { @@ -451,18 +451,34 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem return Futures.immediateFuture(null); }, readResultsProcessingExecutor); - if (query.getRewriteLatestIfDeleted()) { - ListenableFuture savedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { - if (isRemove) { - return getNewLatestEntryFuture(entityId, query); + final SimpleListenableFuture resultFuture = new SimpleListenableFuture<>(); + Futures.addCallback(removedLatestFuture, new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + if (query.getRewriteLatestIfDeleted()) { + ListenableFuture savedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { + if (isRemove) { + return getNewLatestEntryFuture(entityId, query); + } + return Futures.immediateFuture(null); + }, readResultsProcessingExecutor); + + try { + resultFuture.set(savedLatestFuture.get()); + } catch (InterruptedException | ExecutionException e) { + log.warn("Could not get latest saved value for [{}], {}", entityId, query.getKey(), e); + } + } else { + resultFuture.set(null); } - return Futures.immediateFuture(null); - }, readResultsProcessingExecutor); + } - return Futures.transformAsync(Futures.allAsList(Arrays.asList(savedLatestFuture, removedLatestFuture)), - list -> Futures.immediateFuture(null), readResultsProcessingExecutor); - } - return removedLatestFuture; + @Override + public void onFailure(Throwable t) { + log.warn("[{}] Failed to process remove of the latest value", entityId, t); + } + }); + return resultFuture; } private ListenableFuture getNewLatestEntryFuture(EntityId entityId, DeleteTsKvQuery query) { diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java index 88f4d84eb6..81de40ac82 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java @@ -152,7 +152,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { } @Test - public void testDeleteDeviceTsData() throws Exception { + public void testDeleteDeviceTsDataWithoutOverwritingLatest() throws Exception { DeviceId deviceId = new DeviceId(UUIDs.timeBased()); saveEntries(deviceId, 10000); @@ -171,6 +171,26 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { Assert.assertEquals(null, latest.get(0).getValueAsString()); } + @Test + public void testDeleteDeviceTsDataWithOverwritingLatest() throws Exception { + DeviceId deviceId = new DeviceId(UUIDs.timeBased()); + + saveEntries(deviceId, 10000); + saveEntries(deviceId, 20000); + saveEntries(deviceId, 30000); + saveEntries(deviceId, 40000); + + tsService.remove(deviceId, Collections.singletonList( + new BaseDeleteTsKvQuery(STRING_KEY, 25000, 45000, true))).get(); + + List list = tsService.findAll(deviceId, Collections.singletonList( + new BaseReadTsKvQuery(STRING_KEY, 5000, 45000, 10000, 10, Aggregation.NONE))).get(); + Assert.assertEquals(2, list.size()); + + List latest = tsService.findLatest(deviceId, Collections.singletonList(STRING_KEY)).get(); + Assert.assertEquals(20000, latest.get(0).getTs()); + } + @Test public void testFindDeviceTsData() throws Exception { DeviceId deviceId = new DeviceId(UUIDs.timeBased()); From f6fccebd52d4dae21f0a936a14f96b044ee49cb5 Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Tue, 30 Oct 2018 15:35:42 +0200 Subject: [PATCH 06/11] UI fixes --- ui/src/app/common/thirdparty-fix.js | 264 ++++++++++++++++++ ui/src/app/dashboard/dashboard.tpl.html | 1 + .../app/entity-view/entity-view.directive.js | 24 +- ui/src/app/locale/locale.constant-en_US.json | 1 + 4 files changed, 281 insertions(+), 9 deletions(-) diff --git a/ui/src/app/common/thirdparty-fix.js b/ui/src/app/common/thirdparty-fix.js index b00a7822c5..af4222c140 100644 --- a/ui/src/app/common/thirdparty-fix.js +++ b/ui/src/app/common/thirdparty-fix.js @@ -15,10 +15,13 @@ */ import tinycolor from 'tinycolor2'; +import moment from 'moment'; export default angular.module('thingsboard.thirdpartyFix', []) .factory('Fullscreen', Fullscreen) .factory('$mdColorPicker', mdColorPicker) + .provider('$mdpDatePicker', mdpDatePicker) + .provider('$mdpTimePicker', mdpTimePicker) .name; /*@ngInject*/ @@ -193,3 +196,264 @@ function mdColorPicker($q, $mdDialog, mdColorPickerHistory) { /* eslint-enable angular/definedundefined */ } + +function DatePickerCtrl($scope, $mdDialog, $mdMedia, $timeout, currentDate, options) { + var self = this; + + this.date = moment(currentDate); + this.minDate = options.minDate && moment(options.minDate).isValid() ? moment(options.minDate) : null; + this.maxDate = options.maxDate && moment(options.maxDate).isValid() ? moment(options.maxDate) : null; + this.displayFormat = options.displayFormat || "ddd, MMM DD"; + this.dateFilter = angular.isFunction(options.dateFilter) ? options.dateFilter : null; + this.selectingYear = false; + + // validate min and max date + if (this.minDate && this.maxDate) { + if (this.maxDate.isBefore(this.minDate)) { + this.maxDate = moment(this.minDate).add(1, 'days'); + } + } + + if (this.date) { + // check min date + if (this.minDate && this.date.isBefore(this.minDate)) { + this.date = moment(this.minDate); + } + + // check max date + if (this.maxDate && this.date.isAfter(this.maxDate)) { + this.date = moment(this.maxDate); + } + } + + this.yearItems = { + currentIndex_: 0, + PAGE_SIZE: 5, + START: (self.minDate ? self.minDate.year() : 1900), + END: (self.maxDate ? self.maxDate.year() : 0), + getItemAtIndex: function(index) { + if(this.currentIndex_ < index) + this.currentIndex_ = index; + + return this.START + index; + }, + getLength: function() { + return Math.min( + this.currentIndex_ + Math.floor(this.PAGE_SIZE / 2), + Math.abs(this.START - this.END) + 1 + ); + } + }; + + $scope.$mdMedia = $mdMedia; + $scope.year = this.date.year(); + + this.selectYear = function(year) { + self.date.year(year); + $scope.year = year; + self.selectingYear = false; + self.animate(); + }; + + this.showYear = function() { + self.yearTopIndex = (self.date.year() - self.yearItems.START) + Math.floor(self.yearItems.PAGE_SIZE / 2); + self.yearItems.currentIndex_ = (self.date.year() - self.yearItems.START) + 1; + self.selectingYear = true; + }; + + this.showCalendar = function() { + self.selectingYear = false; + }; + + this.cancel = function() { + $mdDialog.cancel(); + }; + + this.confirm = function() { + var date = this.date; + + if (this.minDate && this.date.isBefore(this.minDate)) { + date = moment(this.minDate); + } + + if (this.maxDate && this.date.isAfter(this.maxDate)) { + date = moment(this.maxDate); + } + + $mdDialog.hide(date.toDate()); + }; + + this.animate = function() { + self.animating = true; + $timeout(angular.noop).then(function() { + self.animating = false; + }) + }; +} + +/*@ngInject*/ +function mdpDatePicker() { + var LABEL_OK = "OK", + LABEL_CANCEL = "Cancel", + DISPLAY_FORMAT = "ddd, MMM DD"; + + this.setDisplayFormat = function(format) { + DISPLAY_FORMAT = format; + }; + + this.setOKButtonLabel = function(label) { + LABEL_OK = label; + }; + + this.setCancelButtonLabel = function(label) { + LABEL_CANCEL = label; + }; + + /*@ngInject*/ + this.$get = function($mdDialog) { + var datePicker = function(currentDate, options) { + if (!angular.isDate(currentDate)) currentDate = Date.now(); + if (!angular.isObject(options)) options = {}; + + options.displayFormat = DISPLAY_FORMAT; + + return $mdDialog.show({ + controller: ['$scope', '$mdDialog', '$mdMedia', '$timeout', 'currentDate', 'options', DatePickerCtrl], + controllerAs: 'datepicker', + clickOutsideToClose: true, + template: '' + + '' + + '
' + + '' + + '{{ datepicker.date.format(\'YYYY\') }}' + + '{{ datepicker.date.format(datepicker.displayFormat) }} ' + + '' + + '
' + + '
' + + '
' + + '' + + '
' + + '{{ item }}' + + '
' + + '' + + '
' + + '' + + '' + + '' + + '' + LABEL_CANCEL + '' + + '' + LABEL_OK + '' + + '' + + '
' + + '' + + '', + targetEvent: options.targetEvent, + locals: { + currentDate: currentDate, + options: options + }, + multiple: true + }); + }; + + return datePicker; + }; + +} + +function TimePickerCtrl($scope, $mdDialog, time, autoSwitch, $mdMedia) { + var self = this; + this.VIEW_HOURS = 1; + this.VIEW_MINUTES = 2; + this.currentView = this.VIEW_HOURS; + this.time = moment(time); + this.autoSwitch = !!autoSwitch; + + this.clockHours = parseInt(this.time.format("h")); + this.clockMinutes = parseInt(this.time.minutes()); + + $scope.$mdMedia = $mdMedia; + + this.switchView = function() { + self.currentView = self.currentView == self.VIEW_HOURS ? self.VIEW_MINUTES : self.VIEW_HOURS; + }; + + this.setAM = function() { + if(self.time.hours() >= 12) + self.time.hour(self.time.hour() - 12); + }; + + this.setPM = function() { + if(self.time.hours() < 12) + self.time.hour(self.time.hour() + 12); + }; + + this.cancel = function() { + $mdDialog.cancel(); + }; + + this.confirm = function() { + $mdDialog.hide(this.time.toDate()); + }; +} + +/*@ngInject*/ +function mdpTimePicker() { + var LABEL_OK = "OK", + LABEL_CANCEL = "Cancel"; + + this.setOKButtonLabel = function(label) { + LABEL_OK = label; + }; + + this.setCancelButtonLabel = function(label) { + LABEL_CANCEL = label; + }; + + /*@ngInject*/ + this.$get = function($mdDialog) { + var timePicker = function(time, options) { + if(!angular.isDate(time)) time = Date.now(); + if (!angular.isObject(options)) options = {}; + + return $mdDialog.show({ + controller: ['$scope', '$mdDialog', 'time', 'autoSwitch', '$mdMedia', TimePickerCtrl], + controllerAs: 'timepicker', + clickOutsideToClose: true, + template: '' + + '' + + '' + + '
' + + '{{ timepicker.time.format("h") }}:' + + '{{ timepicker.time.format("mm") }}' + + '
' + + '
' + + 'AM' + + 'PM' + + '
' + + '' + + '
' + + '
' + + '' + + '' + + '
' + + + '' + + '' + + '' + LABEL_CANCEL + '' + + '' + LABEL_OK + '' + + '' + + '
' + + '' + + '', + targetEvent: options.targetEvent, + locals: { + time: time, + autoSwitch: options.autoSwitch + }, + multiple: true + }); + }; + + return timePicker; + }; +} diff --git a/ui/src/app/dashboard/dashboard.tpl.html b/ui/src/app/dashboard/dashboard.tpl.html index 79653a42c5..a1bb2688c0 100644 --- a/ui/src/app/dashboard/dashboard.tpl.html +++ b/ui/src/app/dashboard/dashboard.tpl.html @@ -74,6 +74,7 @@ settings diff --git a/ui/src/app/entity-view/entity-view.directive.js b/ui/src/app/entity-view/entity-view.directive.js index 25377f4cb4..761930e615 100644 --- a/ui/src/app/entity-view/entity-view.directive.js +++ b/ui/src/app/entity-view/entity-view.directive.js @@ -98,8 +98,8 @@ export default function EntityViewDirective($q, $compile, $templateCache, $filte if (newDate.getTime() > scope.maxStartTimeMs) { scope.startTimeMs = angular.copy(scope.maxStartTimeMs); } - updateMinMaxDates(); } + updateMinMaxDates(); }); scope.$watch('endTimeMs', function (newDate) { @@ -107,18 +107,24 @@ export default function EntityViewDirective($q, $compile, $templateCache, $filte if (newDate.getTime() < scope.minEndTimeMs) { scope.endTimeMs = angular.copy(scope.minEndTimeMs); } - updateMinMaxDates(); } + updateMinMaxDates(); }); function updateMinMaxDates() { - if (scope.endTimeMs) { - scope.maxStartTimeMs = angular.copy(new Date(scope.endTimeMs.getTime())); - scope.entityView.endTimeMs = scope.endTimeMs.getTime(); - } - if (scope.startTimeMs) { - scope.minEndTimeMs = angular.copy(new Date(scope.startTimeMs.getTime())); - scope.entityView.startTimeMs = scope.startTimeMs.getTime(); + if (scope.entityView) { + if (scope.endTimeMs) { + scope.maxStartTimeMs = angular.copy(new Date(scope.endTimeMs.getTime())); + scope.entityView.endTimeMs = scope.endTimeMs.getTime(); + } else { + scope.entityView.endTimeMs = 0; + } + if (scope.startTimeMs) { + scope.minEndTimeMs = angular.copy(new Date(scope.startTimeMs.getTime())); + scope.entityView.startTimeMs = scope.startTimeMs.getTime(); + } else { + scope.entityView.startTimeMs = 0; + } } } diff --git a/ui/src/app/locale/locale.constant-en_US.json b/ui/src/app/locale/locale.constant-en_US.json index b3c89c7729..30e1eb119b 100644 --- a/ui/src/app/locale/locale.constant-en_US.json +++ b/ui/src/app/locale/locale.constant-en_US.json @@ -774,6 +774,7 @@ }, "entity-view": { "entity-view": "Entity View", + "entity-view-required": "Entity view is required.", "entity-views": "Entity Views", "management": "Entity View management", "view-entity-views": "View Entity Views", From d96f6c8c3426115b0d254f2e5c76d781ab50db19 Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Wed, 31 Oct 2018 10:52:22 +0200 Subject: [PATCH 07/11] Update README.md --- msa/tb/README.md | 35 ++++++++++++++++++++++++++--------- 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/msa/tb/README.md b/msa/tb/README.md index 8ed84c5e1f..9d15b37587 100644 --- a/msa/tb/README.md +++ b/msa/tb/README.md @@ -25,11 +25,28 @@ Where: - `-v ~/.mytb-data:/data` - mounts the host's dir `~/.mytb-data` to ThingsBoard DataBase data directory - `--name mytb` - friendly local name of this machine - `thingsboard/tb` - docker image, can be also `thingsboard/tb-postgres` or `thingsboard/tb-cassandra` - -After executing this command you can open `http://{your-host-ip}:9090` in you browser (for ex. `http://localhost:9090`). You should see ThingsBoard login page. +- **NOTE**: **Windows** users should use docker managed volume instead of host's dir. Create docker volume (for ex. `mytb-data`) before executing `docker run` command: + + ``` + $ docker create volume mytb-data + ``` + + After you can execute docker run command using `mytb-data` volume instead of `~/.mytb-data`. + + In order to get access to necessary resources from external IP/Host on **Windows** machine, please execute the following commands: + ``` + # HTTP port + $ VBoxManage controlvm "default" natpf1 "tcp-port9090,tcp,,9090,,9090" + # MQTT port + $ VBoxManage controlvm "default" natpf1 "tcp-port1883,tcp,,1883,,1883" + # CoAP port + $ VBoxManage controlvm "default" natpf1 "tcp-port5683,tcp,,5683,,5683" + ``` + +After executing `docker run` command you can open `http://{your-host-ip}:9090` in you browser (for ex. `http://localhost:9090`). You should see ThingsBoard login page. Use the following default credentials: -- **Systen Administrator**: sysadmin@thingsboard.org / sysadmin +- **System Administrator**: sysadmin@thingsboard.org / sysadmin - **Tenant Administrator**: tenant@thingsboard.org / tenant - **Customer User**: customer@thingsboard.org / customer @@ -39,21 +56,21 @@ You can detach from session terminal with `Ctrl-p` `Ctrl-q` - the container will To reattach to the terminal (to see ThingsBoard logs) run: -` +``` $ docker attach mytb -` +``` To stop the container: -` +``` $ docker stop mytb -` +``` To start the container: -` +``` $ docker start mytb -` +``` ## Upgrading From f875bcd65045c5d4235b71073ce974665a595f75 Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Wed, 31 Oct 2018 10:53:23 +0200 Subject: [PATCH 08/11] Update README.md --- msa/tb/README.md | 3 --- 1 file changed, 3 deletions(-) diff --git a/msa/tb/README.md b/msa/tb/README.md index 9d15b37587..bb9fa0f08d 100644 --- a/msa/tb/README.md +++ b/msa/tb/README.md @@ -35,11 +35,8 @@ Where: In order to get access to necessary resources from external IP/Host on **Windows** machine, please execute the following commands: ``` - # HTTP port $ VBoxManage controlvm "default" natpf1 "tcp-port9090,tcp,,9090,,9090" - # MQTT port $ VBoxManage controlvm "default" natpf1 "tcp-port1883,tcp,,1883,,1883" - # CoAP port $ VBoxManage controlvm "default" natpf1 "tcp-port5683,tcp,,5683,,5683" ``` From 27dec3faa0b06f6ead073b0fd3b885f3f617c506 Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Wed, 31 Oct 2018 10:56:21 +0200 Subject: [PATCH 09/11] Update README.md --- msa/tb/README.md | 26 ++++++++++++-------------- 1 file changed, 12 insertions(+), 14 deletions(-) diff --git a/msa/tb/README.md b/msa/tb/README.md index bb9fa0f08d..e4b7400b50 100644 --- a/msa/tb/README.md +++ b/msa/tb/README.md @@ -25,20 +25,18 @@ Where: - `-v ~/.mytb-data:/data` - mounts the host's dir `~/.mytb-data` to ThingsBoard DataBase data directory - `--name mytb` - friendly local name of this machine - `thingsboard/tb` - docker image, can be also `thingsboard/tb-postgres` or `thingsboard/tb-cassandra` -- **NOTE**: **Windows** users should use docker managed volume instead of host's dir. Create docker volume (for ex. `mytb-data`) before executing `docker run` command: - - ``` - $ docker create volume mytb-data - ``` - - After you can execute docker run command using `mytb-data` volume instead of `~/.mytb-data`. - - In order to get access to necessary resources from external IP/Host on **Windows** machine, please execute the following commands: - ``` - $ VBoxManage controlvm "default" natpf1 "tcp-port9090,tcp,,9090,,9090" - $ VBoxManage controlvm "default" natpf1 "tcp-port1883,tcp,,1883,,1883" - $ VBoxManage controlvm "default" natpf1 "tcp-port5683,tcp,,5683,,5683" - ``` + +> **NOTE**: **Windows** users should use docker managed volume instead of host's dir. Create docker volume (for ex. `mytb-data`) before executing `docker run` command: +> ``` +> $ docker create volume mytb-data +> ``` +> After you can execute docker run command using `mytb-data` volume instead of `~/.mytb-data`. +> In order to get access to necessary resources from external IP/Host on **Windows** machine, please execute the following commands: +> ``` +> $ VBoxManage controlvm "default" natpf1 "tcp-port9090,tcp,,9090,,9090" +> $ VBoxManage controlvm "default" natpf1 "tcp-port1883,tcp,,1883,,1883" +> $ VBoxManage controlvm "default" natpf1 "tcp-port5683,tcp,,5683,,5683" +> ``` After executing `docker run` command you can open `http://{your-host-ip}:9090` in you browser (for ex. `http://localhost:9090`). You should see ThingsBoard login page. Use the following default credentials: From 2553c8c57b20f3eddd50bf5080fc738836a71689 Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Wed, 31 Oct 2018 13:06:04 +0200 Subject: [PATCH 10/11] Revert "[WIP]Help refactor code solid patterns" --- .../client/tools/MqttSslClient.java | 106 +++++------------- 1 file changed, 29 insertions(+), 77 deletions(-) diff --git a/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java b/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java index 339f26ded8..dd27e44a1d 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java +++ b/tools/src/main/java/org/thingsboard/client/tools/MqttSslClient.java @@ -50,86 +50,38 @@ public class MqttSslClient { File ksFile = new File(ksUrl.toURI()); URL tsUrl = Resources.getResource(KEY_STORE_FILE); File tsFile = new File(tsUrl.toURI()); - char[] ksPwd = setKeyPassword(); - - KeyStore trustStore = setKeyStore(tsFile, ksPwd); - - TrustManagerFactory tmf = setTrustManagerFactory(trustStore); - - KeyStore ks = setKeyStore2(ksFile, ksPwd); - - KeyManagerFactory kmf = setKeyManagerFactory(ks); - - SSLContext sslContext = setSSLContext(kmf, tmf); - - MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(sslContext); - - MqttAsyncClient mqttAsyncClient = setMqttAsyncClient(mqttConnectOptions); - - setMessage(mqttAsyncClient, "v1/devices/me/telemetry", "{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4}"); + TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + + KeyStore trustStore = KeyStore.getInstance(JKS); + char[] ksPwd = new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x73, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; + trustStore.load(new FileInputStream(tsFile), ksPwd); + tmf.init(trustStore); + KeyStore ks = KeyStore.getInstance(JKS); + + ks.load(new FileInputStream(ksFile), ksPwd); + KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + char[] clientPwd = new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x65, 0x79, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; + kmf.init(ks, clientPwd); + + KeyManager[] km = kmf.getKeyManagers(); + TrustManager[] tm = tmf.getTrustManagers(); + SSLContext sslContext = SSLContext.getInstance(TLS); + sslContext.init(km, tm, null); + + MqttConnectOptions options = new MqttConnectOptions(); + options.setSocketFactory(sslContext.getSocketFactory()); + MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, CLIENT_ID); + client.connect(options); + Thread.sleep(3000); + MqttMessage message = new MqttMessage(); + message.setPayload("{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4}".getBytes()); + client.publish("v1/devices/me/telemetry", message); + client.disconnect(); + log.info("Disconnected"); + System.exit(0); } catch (Exception e) { log.error("Unexpected exception occurred in MqttSslClient", e); } } - - private static char[] setKeyPassword(){ - return new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x73, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; - } - - private static KeyStore setKeyStore(File tsFile, char[] ksPwd) throws Exception{ - KeyStore trustStore = KeyStore.getInstance(JKS); - trustStore.load(new FileInputStream(tsFile), ksPwd); - return trustStore; - } - - private static TrustManagerFactory setTrustManagerFactory(KeyStore trustStore) throws Exception { - TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); - tmf.init(trustStore); - return tmf; - } - - private static KeyStore setKeyStore2(File ksFile, char[] ksPwd)throws Exception{ - KeyStore ks = KeyStore.getInstance(JKS); - ks.load(new FileInputStream(ksFile), ksPwd); - return ks; - } - - private static KeyManagerFactory setKeyManagerFactory(KeyStore ks) throws Exception { - KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); - char[] clientPwd = new char[]{0x63, 0x6C, 0x69, 0x65, 0x6E, 0x74, 0x5F, 0x6B, 0x65, 0x79, 0x5F, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6F, 0x72, 0x64}; - kmf.init(ks, clientPwd); - return kmf; - } - - private static SSLContext setSSLContext(KeyManagerFactory kmf, TrustManagerFactory tmf) throws Exception { - KeyManager[] km = kmf.getKeyManagers(); - TrustManager[] tm = tmf.getTrustManagers(); - SSLContext sslContext = SSLContext.getInstance(TLS); - sslContext.init(km, tm, null); - return sslContext; - } - - private static MqttConnectOptions setMqttConnectOptions(SSLContext sslContext){ - MqttConnectOptions options = new MqttConnectOptions(); - options.setSocketFactory(sslContext.getSocketFactory()); - return options; - } - - private static MqttAsyncClient setMqttAsyncClient(MqttConnectOptions options) throws Exception{ - MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, CLIENT_ID); - client.connect(options); - Thread.sleep(3000); - return client; - } - - private static void setMessage(MqttAsyncClient client, String topic, String payload) throws Exception { - MqttMessage message = new MqttMessage(); - message.setPayload(payload.getBytes()); - client.publish(topic, message); - client.disconnect(); - log.info("Disconnected"); - System.exit(0); - } - } \ No newline at end of file From 32dd7f0437053fdbb42227ff939dc9f2e7be17d1 Mon Sep 17 00:00:00 2001 From: Viacheslav Kukhtyn Date: Wed, 31 Oct 2018 13:07:43 +0200 Subject: [PATCH 11/11] Black box tests update --- msa/black-box-tests/README.md | 21 ++++++++++++ .../pom.xml | 10 +++--- .../server/msa/AbstractContainerTest.java | 0 .../server/msa/ContainerTestSuite.java | 9 +++-- .../org/thingsboard/server/msa/WsClient.java | 33 ++++++++++++++++--- .../msa/connectivity/HttpClientTest.java | 8 ++--- .../msa/connectivity/MqttClientTest.java | 25 +++++++------- .../server/msa/mapper/AttributesResponse.java | 0 .../msa/mapper/WsTelemetryResponse.java | 0 .../RpcResponseRuleChainMetadata.json | 0 msa/integration-tests/README.md | 18 ---------- msa/pom.xml | 4 +-- 12 files changed, 80 insertions(+), 48 deletions(-) create mode 100644 msa/black-box-tests/README.md rename msa/{integration-tests => black-box-tests}/pom.xml (92%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java (100%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java (80%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/WsClient.java (52%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java (90%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java (95%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java (100%) rename msa/{integration-tests => black-box-tests}/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java (100%) rename msa/{integration-tests => black-box-tests}/src/test/resources/RpcResponseRuleChainMetadata.json (100%) delete mode 100644 msa/integration-tests/README.md diff --git a/msa/black-box-tests/README.md b/msa/black-box-tests/README.md new file mode 100644 index 0000000000..c26d9c5fc6 --- /dev/null +++ b/msa/black-box-tests/README.md @@ -0,0 +1,21 @@ + +## Black box tests execution +To run the black box tests with using Docker, the local Docker images of Thingsboard's microservices should be built.
+- Build the local Docker images in the directory with the Thingsboard's main [pom.xml](./../../pom.xml): + + mvn clean install -Ddockerfile.skip=false +- Verify that the new local images were built: + + docker image ls +As result, in REPOSITORY column, next images should be present: + + thingsboard/tb-coap-transport + thingsboard/tb-http-transport + thingsboard/tb-mqtt-transport + thingsboard/tb-node + thingsboard/tb-web-ui + thingsboard/tb-js-executor + +- Run the black box tests in the [msa/black-box-tests](../black-box-tests) directory: + + mvn clean install -DblackBoxTests.skip=false \ No newline at end of file diff --git a/msa/integration-tests/pom.xml b/msa/black-box-tests/pom.xml similarity index 92% rename from msa/integration-tests/pom.xml rename to msa/black-box-tests/pom.xml index a9c67ccf93..2af4d2c14e 100644 --- a/msa/integration-tests/pom.xml +++ b/msa/black-box-tests/pom.xml @@ -25,16 +25,16 @@ msa org.thingsboard.msa - integration-tests + black-box-tests - ThingsBoard Integration Tests + ThingsBoard Black Box Tests https://thingsboard.io - Project for ThingsBoard integration tests with using Docker + Project for ThingsBoard black box testing with using Docker UTF-8 ${basedir}/../.. - true + true 1.9.1 1.3.9 4.5.6 @@ -95,7 +95,7 @@ **/*TestSuite.java - ${integrationTests.skip} + ${blackBoxTests.skip} diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java similarity index 100% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java similarity index 80% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java index 0629960b24..495fd94d2e 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java @@ -22,15 +22,18 @@ import org.testcontainers.containers.DockerComposeContainer; import org.testcontainers.containers.wait.strategy.Wait; import java.io.File; +import java.time.Duration; @RunWith(ClasspathSuite.class) -@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*"}) +@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*Test"}) public class ContainerTestSuite { @ClassRule - public static DockerComposeContainer composeContainer = new DockerComposeContainer(new File("./../docker/docker-compose.yml")) + public static DockerComposeContainer composeContainer = new DockerComposeContainer( + new File("./../../docker/docker-compose.yml"), + new File("./../../docker/docker-compose.postgres.yml")) .withPull(false) .withLocalCompose(true) .withTailChildContainers(true) - .withExposedService("tb-web-ui1", 8080, Wait.forHttp("/login")); + .withExposedService("tb-web-ui1", 8080, Wait.forHttp("/login").withStartupTimeout(Duration.ofSeconds(120))); } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/WsClient.java similarity index 52% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/WsClient.java index 5ef238f8aa..a9835ed618 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/WsClient.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/WsClient.java @@ -15,13 +15,23 @@ */ package org.thingsboard.server.msa; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; import org.java_websocket.client.WebSocketClient; import org.java_websocket.handshake.ServerHandshake; +import org.thingsboard.server.msa.mapper.WsTelemetryResponse; +import java.io.IOException; import java.net.URI; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +@Slf4j public class WsClient extends WebSocketClient { - private String message; + private static final ObjectMapper mapper = new ObjectMapper(); + private WsTelemetryResponse message; + + private CountDownLatch latch = new CountDownLatch(1);; public WsClient(URI serverUri) { super(serverUri); @@ -33,11 +43,20 @@ public class WsClient extends WebSocketClient { @Override public void onMessage(String message) { - this.message = message; + try { + WsTelemetryResponse response = mapper.readValue(message, WsTelemetryResponse.class); + if (!response.getData().isEmpty()) { + this.message = response; + latch.countDown(); + } + } catch (IOException e) { + log.error("ws message can't be read"); + } } @Override public void onClose(int code, String reason, boolean remote) { + log.info("ws is closed, due to [{}]", reason); } @Override @@ -45,7 +64,13 @@ public class WsClient extends WebSocketClient { ex.printStackTrace(); } - public String getLastMessage() { - return this.message; + public WsTelemetryResponse getLastMessage() { + try { + latch.await(10, TimeUnit.SECONDS); + return this.message; + } catch (InterruptedException e) { + log.error("Timeout, ws message wasn't received"); + } + return null; } } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java similarity index 90% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java index f8e1041252..a6e89de185 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/HttpClientTest.java @@ -25,12 +25,10 @@ import org.thingsboard.server.msa.AbstractContainerTest; import org.thingsboard.server.msa.WsClient; import org.thingsboard.server.msa.mapper.WsTelemetryResponse; -import java.util.concurrent.TimeUnit; - public class HttpClientTest extends AbstractContainerTest { @Test - public void telemetryUpdate() throws Exception { + public void telemetryUpload() throws Exception { restClient.login("tenant@thingsboard.org", "tenant"); Device device = createDevice("http_"); @@ -43,8 +41,8 @@ public class HttpClientTest extends AbstractContainerTest { ResponseEntity.class, deviceCredentials.getCredentialsId()); Assert.assertTrue(deviceTelemetryResponse.getStatusCode().is2xxSuccessful()); - TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); + wsClient.closeBlocking(); Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"), actualLatestTelemetry.getLatestValues().keySet()); diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java similarity index 95% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java index eae98b9f1e..d889d2c52e 100644 --- a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java @@ -23,7 +23,9 @@ import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.JsonObject; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.handler.codec.mqtt.MqttQoS; import lombok.Data; +import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.RandomStringUtils; import org.junit.*; import org.springframework.core.ParameterizedTypeReference; @@ -50,6 +52,7 @@ import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.*; +@Slf4j public class MqttClientTest extends AbstractContainerTest { @Test @@ -61,8 +64,8 @@ public class MqttClientTest extends AbstractContainerTest { WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); MqttClient mqttClient = getMqttClient(deviceCredentials, null); mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload().toString().getBytes())); - TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); + wsClient.closeBlocking(); Assert.assertEquals(4, actualLatestTelemetry.getData().size()); Assert.assertEquals(Sets.newHashSet("booleanKey", "stringKey", "doubleKey", "longKey"), @@ -87,8 +90,8 @@ public class MqttClientTest extends AbstractContainerTest { WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); MqttClient mqttClient = getMqttClient(deviceCredentials, null); mqttClient.publish("v1/devices/me/telemetry", Unpooled.wrappedBuffer(createPayload(ts).toString().getBytes())); - TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); + wsClient.closeBlocking(); Assert.assertEquals(4, actualLatestTelemetry.getData().size()); Assert.assertEquals(getExpectedLatestValues(ts), actualLatestTelemetry.getLatestValues()); @@ -116,8 +119,8 @@ public class MqttClientTest extends AbstractContainerTest { clientAttributes.addProperty("attr3", 42.0); clientAttributes.addProperty("attr4", 73); mqttClient.publish("v1/devices/me/attributes", Unpooled.wrappedBuffer(clientAttributes.toString().getBytes())); - TimeUnit.SECONDS.sleep(1); - WsTelemetryResponse actualLatestTelemetry = mapper.readValue(wsClient.getLastMessage(), WsTelemetryResponse.class); + WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); + wsClient.closeBlocking(); Assert.assertEquals(4, actualLatestTelemetry.getData().size()); Assert.assertEquals(Sets.newHashSet("attr1", "attr2", "attr3", "attr4"), @@ -157,7 +160,7 @@ public class MqttClientTest extends AbstractContainerTest { Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful()); // Subscribe to attributes response - mqttClient.on("v1/devices/me/attributes/response/+", listener); + mqttClient.on("v1/devices/me/attributes/response/+", listener, MqttQoS.AT_LEAST_ONCE); // Request attributes JsonObject request = new JsonObject(); request.addProperty("clientKeys", "clientAttr"); @@ -183,7 +186,7 @@ public class MqttClientTest extends AbstractContainerTest { MqttMessageListener listener = new MqttMessageListener(); MqttClient mqttClient = getMqttClient(deviceCredentials, listener); - mqttClient.on("v1/devices/me/attributes", listener); + mqttClient.on("v1/devices/me/attributes", listener, MqttQoS.AT_LEAST_ONCE); String sharedAttributeName = "sharedAttr"; @@ -226,13 +229,12 @@ public class MqttClientTest extends AbstractContainerTest { MqttMessageListener listener = new MqttMessageListener(); MqttClient mqttClient = getMqttClient(deviceCredentials, listener); - mqttClient.on("v1/devices/me/rpc/request/+", listener); + mqttClient.on("v1/devices/me/rpc/request/+", listener, MqttQoS.AT_LEAST_ONCE); // Send an RPC from the server JsonObject serverRpcPayload = new JsonObject(); serverRpcPayload.addProperty("method", "getValue"); serverRpcPayload.addProperty("params", true); - serverRpcPayload.addProperty("timeout", 1000); ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); ListenableFuture future = service.submit(() -> { try { @@ -271,7 +273,7 @@ public class MqttClientTest extends AbstractContainerTest { MqttMessageListener listener = new MqttMessageListener(); MqttClient mqttClient = getMqttClient(deviceCredentials, listener); - mqttClient.on("v1/devices/me/rpc/request/+", listener); + mqttClient.on("v1/devices/me/rpc/request/+", listener, MqttQoS.AT_LEAST_ONCE); // Get the default rule chain id to make it root again after test finished RuleChainId defaultRuleChainId = getDefaultRuleChainId(); @@ -377,6 +379,7 @@ public class MqttClientTest extends AbstractContainerTest { @Override public void onMessage(String topic, ByteBuf message) { + log.info("MQTT message [{}], topic [{}]", message.toString(StandardCharsets.UTF_8), topic); events.add(new MqttEvent(topic, message.toString(StandardCharsets.UTF_8))); } } diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java similarity index 100% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/AttributesResponse.java diff --git a/msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java similarity index 100% rename from msa/integration-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java rename to msa/black-box-tests/src/test/java/org/thingsboard/server/msa/mapper/WsTelemetryResponse.java diff --git a/msa/integration-tests/src/test/resources/RpcResponseRuleChainMetadata.json b/msa/black-box-tests/src/test/resources/RpcResponseRuleChainMetadata.json similarity index 100% rename from msa/integration-tests/src/test/resources/RpcResponseRuleChainMetadata.json rename to msa/black-box-tests/src/test/resources/RpcResponseRuleChainMetadata.json diff --git a/msa/integration-tests/README.md b/msa/integration-tests/README.md deleted file mode 100644 index 93df2bf77f..0000000000 --- a/msa/integration-tests/README.md +++ /dev/null @@ -1,18 +0,0 @@ - -## Integration tests execution -To run the integration tests with using Docker, the local Docker images of Thingsboard's microservices should be built.
-- Build the local Docker images in the directory with the Thingsboard's main [pom.xml](./../../pom.xml): - - mvn clean install -Ddockerfile.skip=false -- Verify that the new local images were built: - - docker image ls -As result, in REPOSITORY column, next images should be present: - - local-maven-build/tb-node - local-maven-build/tb-web-ui - local-maven-build/tb-web-ui - -- Run the integration tests in the [msa/integration-tests](../integration-tests) directory: - - mvn clean install -DintegrationTests.skip=false \ No newline at end of file diff --git a/msa/pom.xml b/msa/pom.xml index 41926608be..5a9d9abd80 100644 --- a/msa/pom.xml +++ b/msa/pom.xml @@ -16,7 +16,7 @@ --> + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 org.thingsboard @@ -41,7 +41,7 @@ web-ui tb-node transport - integration-tests + black-box-tests