Oleg Kolesnik
8 years ago
113 changed files with 2603 additions and 540 deletions
@ -0,0 +1,27 @@ |
|||
/** |
|||
* 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.actors.app; |
|||
|
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
|
|||
public class AppInitMsg implements TbActorMsg { |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.APP_INIT_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
/** |
|||
* 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.actors.device; |
|||
|
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public class DeviceActorCreator extends ContextBasedCreator<DeviceActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final TenantId tenantId; |
|||
private final DeviceId deviceId; |
|||
|
|||
public DeviceActorCreator(ActorSystemContext context, TenantId tenantId, DeviceId deviceId) { |
|||
super(context); |
|||
this.tenantId = tenantId; |
|||
this.deviceId = deviceId; |
|||
} |
|||
|
|||
@Override |
|||
public DeviceActor create() { |
|||
return new DeviceActor(context, tenantId, deviceId); |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* 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.service.executors; |
|||
|
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
@Component |
|||
public class ClusterRpcCallbackExecutorService extends AbstractListeningExecutor { |
|||
|
|||
@Value("${actors.cluster.grpc_callback_thread_pool_size}") |
|||
private int grpcCallbackExecutorThreadPoolSize; |
|||
|
|||
@Override |
|||
protected int getThreadPollSize() { |
|||
return grpcCallbackExecutorThreadPoolSize; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* 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.common.transport.adaptor; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.context.annotation.Configuration; |
|||
|
|||
@Configuration |
|||
@Slf4j |
|||
public class JsonConverterConfig { |
|||
|
|||
@Value("${transport.json.type_cast_enabled:true}") |
|||
public void setIsJsonTypeCastEnabled(boolean jsonTypeCastEnabled) { |
|||
JsonConverter.setTypeCastEnabled(jsonTypeCastEnabled); |
|||
log.info("JSON type cast enabled = {}", jsonTypeCastEnabled); |
|||
} |
|||
} |
|||
@ -1,26 +0,0 @@ |
|||
#!/bin/bash |
|||
# |
|||
# 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. |
|||
# |
|||
|
|||
dirsArray=("./haproxy/certs.d" "./haproxy/letsencrypt" "./tb-node/postgres" "./tb-node/cassandra" "./tb-node/log/tb1" "./tb-node/log/tb2") |
|||
|
|||
for dir in ${dirsArray[@]} |
|||
do |
|||
if [ ! -d "$dir" ]; then |
|||
echo creating dir $dir |
|||
mkdir -p $dir |
|||
fi |
|||
done |
|||
@ -0,0 +1,60 @@ |
|||
# |
|||
# 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. |
|||
# |
|||
|
|||
version: '2.2' |
|||
|
|||
services: |
|||
postgres: |
|||
volumes: |
|||
- postgres-db-volume:/var/lib/postgresql/data |
|||
tb1: |
|||
volumes: |
|||
- tb-log-volume:/var/log/thingsboard |
|||
tb2: |
|||
volumes: |
|||
- tb-log-volume:/var/log/thingsboard |
|||
tb-coap-transport: |
|||
volumes: |
|||
- tb-coap-transport-log-volume:/var/log/tb-coap-transport |
|||
tb-http-transport1: |
|||
volumes: |
|||
- tb-http-transport-log-volume:/var/log/tb-http-transport |
|||
tb-http-transport2: |
|||
volumes: |
|||
- tb-http-transport-log-volume:/var/log/tb-http-transport |
|||
tb-mqtt-transport1: |
|||
volumes: |
|||
- tb-mqtt-transport-log-volume:/var/log/tb-mqtt-transport |
|||
tb-mqtt-transport2: |
|||
volumes: |
|||
- tb-mqtt-transport-log-volume:/var/log/tb-mqtt-transport |
|||
|
|||
volumes: |
|||
postgres-db-volume: |
|||
external: true |
|||
name: ${POSTGRES_DATA_VOLUME} |
|||
tb-log-volume: |
|||
external: true |
|||
name: ${TB_LOG_VOLUME} |
|||
tb-coap-transport-log-volume: |
|||
external: true |
|||
name: ${TB_COAP_TRANSPORT_LOG_VOLUME} |
|||
tb-http-transport-log-volume: |
|||
external: true |
|||
name: ${TB_HTTP_TRANSPORT_LOG_VOLUME} |
|||
tb-mqtt-transport-log-volume: |
|||
external: true |
|||
name: ${TB_MQTT_TRANSPORT_LOG_VOLUME} |
|||
@ -1,8 +1,7 @@ |
|||
|
|||
HTTP_BIND_ADDRESS=0.0.0.0 |
|||
HTTP_BIND_PORT=8080 |
|||
TB_HOST=haproxy |
|||
TB_PORT=8080 |
|||
TB_ENABLE_PROXY=false |
|||
LOGGER_LEVEL=info |
|||
LOG_FOLDER=logs |
|||
LOGGER_FILENAME=tb-web-ui-%DATE%.log |
|||
|
|||
@ -0,0 +1,23 @@ |
|||
|
|||
## Black box tests execution |
|||
To run the black box tests with using Docker, the local Docker images of Thingsboard's microservices should be built. <br /> |
|||
- 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 |
|||
|
|||
|
|||
@ -0,0 +1,110 @@ |
|||
<!-- |
|||
|
|||
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. |
|||
|
|||
--> |
|||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" |
|||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> |
|||
<modelVersion>4.0.0</modelVersion> |
|||
|
|||
<parent> |
|||
<groupId>org.thingsboard</groupId> |
|||
<version>2.2.0-SNAPSHOT</version> |
|||
<artifactId>msa</artifactId> |
|||
</parent> |
|||
<groupId>org.thingsboard.msa</groupId> |
|||
<artifactId>black-box-tests</artifactId> |
|||
|
|||
<name>ThingsBoard Black Box Tests</name> |
|||
<url>https://thingsboard.io</url> |
|||
<description>Project for ThingsBoard black box testing with using Docker</description> |
|||
|
|||
<properties> |
|||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> |
|||
<main.dir>${basedir}/../..</main.dir> |
|||
<blackBoxTests.skip>true</blackBoxTests.skip> |
|||
<testcontainers.version>1.9.1</testcontainers.version> |
|||
<zeroturnaround.version>1.10</zeroturnaround.version> |
|||
<java-websocket.version>1.3.9</java-websocket.version> |
|||
<httpclient.version>4.5.6</httpclient.version> |
|||
</properties> |
|||
|
|||
<dependencies> |
|||
<dependency> |
|||
<groupId>org.testcontainers</groupId> |
|||
<artifactId>testcontainers</artifactId> |
|||
<version>${testcontainers.version}</version> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.zeroturnaround</groupId> |
|||
<artifactId>zt-exec</artifactId> |
|||
<version>${zeroturnaround.version}</version> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.java-websocket</groupId> |
|||
<artifactId>Java-WebSocket</artifactId> |
|||
<version>${java-websocket.version}</version> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.apache.httpcomponents</groupId> |
|||
<artifactId>httpclient</artifactId> |
|||
<version>${httpclient.version}</version> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>io.takari.junit</groupId> |
|||
<artifactId>takari-cpsuite</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>ch.qos.logback</groupId> |
|||
<artifactId>logback-classic</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>com.google.code.gson</groupId> |
|||
<artifactId>gson</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.apache.commons</groupId> |
|||
<artifactId>commons-lang3</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>com.google.guava</groupId> |
|||
<artifactId>guava</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.thingsboard</groupId> |
|||
<artifactId>netty-mqtt</artifactId> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.thingsboard</groupId> |
|||
<artifactId>tools</artifactId> |
|||
</dependency> |
|||
</dependencies> |
|||
|
|||
<build> |
|||
<plugins> |
|||
<plugin> |
|||
<groupId>org.apache.maven.plugins</groupId> |
|||
<artifactId>maven-surefire-plugin</artifactId> |
|||
<configuration> |
|||
<includes> |
|||
<include>**/*TestSuite.java</include> |
|||
</includes> |
|||
<skipTests>${blackBoxTests.skip}</skipTests> |
|||
</configuration> |
|||
</plugin> |
|||
</plugins> |
|||
</build> |
|||
|
|||
</project> |
|||
@ -0,0 +1,206 @@ |
|||
/** |
|||
* 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.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.junit.rules.TestRule; |
|||
import org.junit.rules.TestWatcher; |
|||
import org.junit.runner.Description; |
|||
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.security.cert.X509Certificate; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Random; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractContainerTest { |
|||
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() throws Exception { |
|||
restClient = new RestClient(HTTPS_URL); |
|||
restClient.getRestTemplate().setRequestFactory(getRequestFactoryForSelfSignedCert()); |
|||
} |
|||
|
|||
@Rule |
|||
public TestRule watcher = new TestWatcher() { |
|||
protected void starting(Description description) { |
|||
log.info("================================================="); |
|||
log.info("STARTING TEST: {}" , description.getMethodName()); |
|||
log.info("================================================="); |
|||
} |
|||
|
|||
/** |
|||
* Invoked when a test succeeds |
|||
*/ |
|||
protected void succeeded(Description description) { |
|||
log.info("================================================="); |
|||
log.info("SUCCEEDED TEST: {}" , description.getMethodName()); |
|||
log.info("================================================="); |
|||
} |
|||
|
|||
/** |
|||
* Invoked when a test fails |
|||
*/ |
|||
protected void failed(Throwable e, Description description) { |
|||
log.info("================================================="); |
|||
log.info("FAILED TEST: {}" , description.getMethodName(), e); |
|||
log.info("================================================="); |
|||
} |
|||
}; |
|||
|
|||
protected Device createDevice(String name) { |
|||
return restClient.createDevice(name + RandomStringUtils.randomAlphanumeric(7), "DEFAULT"); |
|||
} |
|||
|
|||
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(property.toString(), cmd); |
|||
wsClient.send(wsRequest.toString()); |
|||
wsClient.waitForFirstReply(); |
|||
return wsClient; |
|||
} |
|||
|
|||
protected Map<String, Long> getExpectedLatestValues(long ts) { |
|||
return ImmutableMap.<String, Long>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<Object> 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<Object> 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; |
|||
} |
|||
|
|||
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<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder |
|||
.<ConnectionSocketFactory>create() |
|||
.register("https", sslSelfSigned) |
|||
.build(); |
|||
|
|||
PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(socketFactoryRegistry); |
|||
CloseableHttpClient httpClient = HttpClients.custom().setConnectionManager(cm).build(); |
|||
return new HttpComponentsClientHttpRequestFactory(httpClient); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,58 @@ |
|||
/** |
|||
* 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.rules.ExternalResource; |
|||
import org.junit.runner.RunWith; |
|||
import org.testcontainers.containers.DockerComposeContainer; |
|||
import org.testcontainers.containers.wait.strategy.Wait; |
|||
import org.testcontainers.utility.Base58; |
|||
|
|||
import java.io.File; |
|||
import java.time.Duration; |
|||
import java.util.Arrays; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
@RunWith(ClasspathSuite.class) |
|||
@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.msa.*Test"}) |
|||
public class ContainerTestSuite { |
|||
|
|||
private static DockerComposeContainer testContainer; |
|||
|
|||
@ClassRule |
|||
public static ThingsBoardDbInstaller installTb = new ThingsBoardDbInstaller(); |
|||
|
|||
@ClassRule |
|||
public static DockerComposeContainer getTestContainer() { |
|||
if (testContainer == null) { |
|||
testContainer = new DockerComposeContainer( |
|||
new File("./../../docker/docker-compose.yml"), |
|||
new File("./../../docker/docker-compose.postgres.yml"), |
|||
new File("./../../docker/docker-compose.postgres.volumes.yml")) |
|||
.withPull(false) |
|||
.withLocalCompose(true) |
|||
.withTailChildContainers(true) |
|||
.withEnv(installTb.getEnv()) |
|||
.withEnv("LOAD_BALANCER_NAME", "") |
|||
.withExposedService("haproxy", 80, Wait.forHttp("/swagger-ui.html").withStartupTimeout(Duration.ofSeconds(120))); |
|||
} |
|||
return testContainer; |
|||
} |
|||
} |
|||
@ -0,0 +1,119 @@ |
|||
/** |
|||
* 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.google.common.base.Splitter; |
|||
import com.google.common.collect.Maps; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.SystemUtils; |
|||
import org.testcontainers.containers.ContainerLaunchException; |
|||
import org.testcontainers.utility.CommandLine; |
|||
import org.zeroturnaround.exec.InvalidExitValueException; |
|||
import org.zeroturnaround.exec.ProcessExecutor; |
|||
import org.zeroturnaround.exec.stream.slf4j.Slf4jStream; |
|||
|
|||
import java.io.File; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Objects; |
|||
import java.util.stream.Stream; |
|||
|
|||
import static com.google.common.base.Preconditions.checkArgument; |
|||
import static com.google.common.base.Preconditions.checkNotNull; |
|||
import static java.util.stream.Collectors.joining; |
|||
|
|||
@Slf4j |
|||
public class DockerComposeExecutor { |
|||
|
|||
String ENV_PROJECT_NAME = "COMPOSE_PROJECT_NAME"; |
|||
String ENV_COMPOSE_FILE = "COMPOSE_FILE"; |
|||
|
|||
private static final String COMPOSE_EXECUTABLE = SystemUtils.IS_OS_WINDOWS ? "docker-compose.exe" : "docker-compose"; |
|||
private static final String DOCKER_EXECUTABLE = SystemUtils.IS_OS_WINDOWS ? "docker.exe" : "docker"; |
|||
|
|||
private final List<File> composeFiles; |
|||
private final String identifier; |
|||
private String cmd = ""; |
|||
private Map<String, String> env = new HashMap<>(); |
|||
|
|||
public DockerComposeExecutor(List<File> composeFiles, String identifier) { |
|||
validateFileList(composeFiles); |
|||
this.composeFiles = composeFiles; |
|||
this.identifier = identifier; |
|||
} |
|||
|
|||
public DockerComposeExecutor withCommand(String cmd) { |
|||
this.cmd = cmd; |
|||
return this; |
|||
} |
|||
|
|||
public DockerComposeExecutor withEnv(Map<String, String> env) { |
|||
this.env = env; |
|||
return this; |
|||
} |
|||
|
|||
public void invokeCompose() { |
|||
// bail out early
|
|||
if (!CommandLine.executableExists(COMPOSE_EXECUTABLE)) { |
|||
throw new ContainerLaunchException("Local Docker Compose not found. Is " + COMPOSE_EXECUTABLE + " on the PATH?"); |
|||
} |
|||
final Map<String, String> environment = Maps.newHashMap(env); |
|||
environment.put(ENV_PROJECT_NAME, identifier); |
|||
final Stream<String> absoluteDockerComposeFilePaths = composeFiles.stream().map(File::getAbsolutePath).map(Objects::toString); |
|||
final String composeFileEnvVariableValue = absoluteDockerComposeFilePaths.collect(joining(File.pathSeparator + "")); |
|||
log.debug("Set env COMPOSE_FILE={}", composeFileEnvVariableValue); |
|||
final File pwd = composeFiles.get(0).getAbsoluteFile().getParentFile().getAbsoluteFile(); |
|||
environment.put(ENV_COMPOSE_FILE, composeFileEnvVariableValue); |
|||
log.info("Local Docker Compose is running command: {}", cmd); |
|||
final List<String> command = Splitter.onPattern(" ").omitEmptyStrings().splitToList(COMPOSE_EXECUTABLE + " " + cmd); |
|||
try { |
|||
new ProcessExecutor().command(command).redirectOutput(Slf4jStream.of(log).asInfo()).redirectError(Slf4jStream.of(log).asError()).environment(environment).directory(pwd).exitValueNormal().executeNoTimeout(); |
|||
log.info("Docker Compose has finished running"); |
|||
} catch (InvalidExitValueException e) { |
|||
throw new ContainerLaunchException("Local Docker Compose exited abnormally with code " + e.getExitValue() + " whilst running command: " + cmd); |
|||
} catch (Exception e) { |
|||
throw new ContainerLaunchException("Error running local Docker Compose command: " + cmd, e); |
|||
} |
|||
} |
|||
|
|||
public void invokeDocker() { |
|||
// bail out early
|
|||
if (!CommandLine.executableExists(DOCKER_EXECUTABLE)) { |
|||
throw new ContainerLaunchException("Local Docker not found. Is " + DOCKER_EXECUTABLE + " on the PATH?"); |
|||
} |
|||
final File pwd = composeFiles.get(0).getAbsoluteFile().getParentFile().getAbsoluteFile(); |
|||
log.info("Local Docker is running command: {}", cmd); |
|||
final List<String> command = Splitter.onPattern(" ").omitEmptyStrings().splitToList(DOCKER_EXECUTABLE + " " + cmd); |
|||
try { |
|||
new ProcessExecutor().command(command).redirectOutput(Slf4jStream.of(log).asInfo()).redirectError(Slf4jStream.of(log).asError()).directory(pwd).exitValueNormal().executeNoTimeout(); |
|||
log.info("Docker has finished running"); |
|||
} catch (InvalidExitValueException e) { |
|||
throw new ContainerLaunchException("Local Docker exited abnormally with code " + e.getExitValue() + " whilst running command: " + cmd); |
|||
} catch (Exception e) { |
|||
throw new ContainerLaunchException("Error running local Docker command: " + cmd, e); |
|||
} |
|||
} |
|||
|
|||
void validateFileList(List<File> composeFiles) { |
|||
checkNotNull(composeFiles); |
|||
checkArgument(!composeFiles.isEmpty(), "No docker compose file have been provided"); |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,132 @@ |
|||
/** |
|||
* 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.rules.ExternalResource; |
|||
import org.testcontainers.utility.Base58; |
|||
|
|||
import java.io.File; |
|||
import java.util.Arrays; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
public class ThingsBoardDbInstaller extends ExternalResource { |
|||
|
|||
private final static String POSTGRES_DATA_VOLUME = "tb-postgres-test-data-volume"; |
|||
private final static String TB_LOG_VOLUME = "tb-log-test-volume"; |
|||
private final static String TB_COAP_TRANSPORT_LOG_VOLUME = "tb-coap-transport-log-test-volume"; |
|||
private final static String TB_HTTP_TRANSPORT_LOG_VOLUME = "tb-http-transport-log-test-volume"; |
|||
private final static String TB_MQTT_TRANSPORT_LOG_VOLUME = "tb-mqtt-transport-log-test-volume"; |
|||
|
|||
private final DockerComposeExecutor dockerCompose; |
|||
|
|||
private final String postgresDataVolume; |
|||
private final String tbLogVolume; |
|||
private final String tbCoapTransportLogVolume; |
|||
private final String tbHttpTransportLogVolume; |
|||
private final String tbMqttTransportLogVolume; |
|||
private final Map<String, String> env; |
|||
|
|||
public ThingsBoardDbInstaller() { |
|||
List<File> composeFiles = Arrays.asList(new File("./../../docker/docker-compose.yml"), |
|||
new File("./../../docker/docker-compose.postgres.yml"), |
|||
new File("./../../docker/docker-compose.postgres.volumes.yml")); |
|||
|
|||
String identifier = Base58.randomString(6).toLowerCase(); |
|||
String project = identifier + Base58.randomString(6).toLowerCase(); |
|||
|
|||
postgresDataVolume = project + "_" + POSTGRES_DATA_VOLUME; |
|||
tbLogVolume = project + "_" + TB_LOG_VOLUME; |
|||
tbCoapTransportLogVolume = project + "_" + TB_COAP_TRANSPORT_LOG_VOLUME; |
|||
tbHttpTransportLogVolume = project + "_" + TB_HTTP_TRANSPORT_LOG_VOLUME; |
|||
tbMqttTransportLogVolume = project + "_" + TB_MQTT_TRANSPORT_LOG_VOLUME; |
|||
|
|||
dockerCompose = new DockerComposeExecutor(composeFiles, project); |
|||
|
|||
env = new HashMap<>(); |
|||
env.put("POSTGRES_DATA_VOLUME", postgresDataVolume); |
|||
env.put("TB_LOG_VOLUME", tbLogVolume); |
|||
env.put("TB_COAP_TRANSPORT_LOG_VOLUME", tbCoapTransportLogVolume); |
|||
env.put("TB_HTTP_TRANSPORT_LOG_VOLUME", tbHttpTransportLogVolume); |
|||
env.put("TB_MQTT_TRANSPORT_LOG_VOLUME", tbMqttTransportLogVolume); |
|||
dockerCompose.withEnv(env); |
|||
} |
|||
|
|||
public Map<String, String> getEnv() { |
|||
return env; |
|||
} |
|||
|
|||
@Override |
|||
protected void before() throws Throwable { |
|||
try { |
|||
|
|||
dockerCompose.withCommand("volume create " + postgresDataVolume); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("volume create " + tbLogVolume); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("volume create " + tbCoapTransportLogVolume); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("volume create " + tbHttpTransportLogVolume); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("volume create " + tbMqttTransportLogVolume); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("up -d redis postgres"); |
|||
dockerCompose.invokeCompose(); |
|||
|
|||
dockerCompose.withCommand("run --no-deps --rm -e INSTALL_TB=true -e LOAD_DEMO=true tb1"); |
|||
dockerCompose.invokeCompose(); |
|||
|
|||
} finally { |
|||
try { |
|||
dockerCompose.withCommand("down -v"); |
|||
dockerCompose.invokeCompose(); |
|||
} catch (Exception e) {} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected void after() { |
|||
copyLogs(tbLogVolume, "./target/tb-logs/"); |
|||
copyLogs(tbCoapTransportLogVolume, "./target/tb-coap-transport-logs/"); |
|||
copyLogs(tbHttpTransportLogVolume, "./target/tb-http-transport-logs/"); |
|||
copyLogs(tbMqttTransportLogVolume, "./target/tb-mqtt-transport-logs/"); |
|||
|
|||
dockerCompose.withCommand("volume rm -f " + postgresDataVolume + " " + tbLogVolume + |
|||
" " + tbCoapTransportLogVolume + " " + tbHttpTransportLogVolume + " " + tbMqttTransportLogVolume); |
|||
dockerCompose.invokeDocker(); |
|||
} |
|||
|
|||
private void copyLogs(String volumeName, String targetDir) { |
|||
File tbLogsDir = new File(targetDir); |
|||
tbLogsDir.mkdirs(); |
|||
|
|||
dockerCompose.withCommand("run -d --rm --name tb-logs-container -v " + volumeName + ":/root alpine tail -f /dev/null"); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("cp tb-logs-container:/root/. "+tbLogsDir.getAbsolutePath()); |
|||
dockerCompose.invokeDocker(); |
|||
|
|||
dockerCompose.withCommand("rm -f tb-logs-container"); |
|||
dockerCompose.invokeDocker(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,92 @@ |
|||
/** |
|||
* 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 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 static final ObjectMapper mapper = new ObjectMapper(); |
|||
private WsTelemetryResponse message; |
|||
|
|||
private volatile boolean firstReplyReceived; |
|||
private CountDownLatch firstReply = new CountDownLatch(1); |
|||
private CountDownLatch latch = new CountDownLatch(1); |
|||
|
|||
WsClient(URI serverUri) { |
|||
super(serverUri); |
|||
} |
|||
|
|||
@Override |
|||
public void onOpen(ServerHandshake serverHandshake) { |
|||
} |
|||
|
|||
@Override |
|||
public void onMessage(String message) { |
|||
if (!firstReplyReceived) { |
|||
firstReplyReceived = true; |
|||
firstReply.countDown(); |
|||
} else { |
|||
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 |
|||
public void onError(Exception ex) { |
|||
ex.printStackTrace(); |
|||
} |
|||
|
|||
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; |
|||
} |
|||
|
|||
void waitForFirstReply() { |
|||
try { |
|||
firstReply.await(10, TimeUnit.SECONDS); |
|||
} catch (InterruptedException e) { |
|||
log.error("Timeout, ws message wasn't received"); |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
} |
|||
@ -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 com.google.common.collect.Sets; |
|||
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.mapper.WsTelemetryResponse; |
|||
|
|||
public class HttpClientTest extends AbstractContainerTest { |
|||
|
|||
@Test |
|||
public void telemetryUpload() throws Exception { |
|||
restClient.login("tenant@thingsboard.org", "tenant"); |
|||
|
|||
Device device = createDevice("http_"); |
|||
DeviceCredentials deviceCredentials = restClient.getCredentials(device.getId()); |
|||
|
|||
WsClient wsClient = subscribeToWebSocket(device.getId(), "LATEST_TELEMETRY", CmdsType.TS_SUB_CMDS); |
|||
ResponseEntity deviceTelemetryResponse = restClient.getRestTemplate() |
|||
.postForEntity(HTTPS_URL + "/api/v1/{credentialsId}/telemetry", |
|||
mapper.readTree(createPayload().toString()), |
|||
ResponseEntity.class, |
|||
deviceCredentials.getCredentialsId()); |
|||
Assert.assertTrue(deviceTelemetryResponse.getStatusCode().is2xxSuccessful()); |
|||
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); |
|||
wsClient.closeBlocking(); |
|||
|
|||
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(HTTPS_URL + "/api/device/" + device.getId()); |
|||
} |
|||
} |
|||
@ -0,0 +1,400 @@ |
|||
/** |
|||
* 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 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 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.junit.rules.TestRule; |
|||
import org.junit.rules.TestWatcher; |
|||
import org.junit.runner.Description; |
|||
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.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.*; |
|||
|
|||
@Slf4j |
|||
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 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())); |
|||
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); |
|||
log.info("Received telemetry: {}", actualLatestTelemetry); |
|||
wsClient.closeBlocking(); |
|||
|
|||
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(HTTPS_URL + "/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 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())); |
|||
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); |
|||
log.info("Received telemetry: {}", actualLatestTelemetry); |
|||
wsClient.closeBlocking(); |
|||
|
|||
Assert.assertEquals(4, actualLatestTelemetry.getData().size()); |
|||
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(HTTPS_URL + "/api/device/" + device.getId()); |
|||
} |
|||
|
|||
@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())); |
|||
WsTelemetryResponse actualLatestTelemetry = wsClient.getLastMessage(); |
|||
log.info("Received telemetry: {}", actualLatestTelemetry); |
|||
wsClient.closeBlocking(); |
|||
|
|||
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, MqttQoS.AT_LEAST_ONCE); |
|||
// 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); |
|||
log.info("Received telemetry: {}", attributes); |
|||
|
|||
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, MqttQoS.AT_LEAST_ONCE); |
|||
|
|||
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, MqttQoS.AT_LEAST_ONCE); |
|||
|
|||
// Send an RPC from the server
|
|||
JsonObject serverRpcPayload = new JsonObject(); |
|||
serverRpcPayload.addProperty("method", "getValue"); |
|||
serverRpcPayload.addProperty("params", true); |
|||
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); |
|||
ListenableFuture<ResponseEntity> 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, MqttQoS.AT_LEAST_ONCE); |
|||
|
|||
// 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(); |
|||
|
|||
TimeUnit.SECONDS.sleep(3); |
|||
// 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<RuleChain> 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<RuleChain> 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<RuleChainMetaData> 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<RuleChain> 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<TextPageData<RuleChain>> ruleChains = restClient.getRestTemplate().exchange( |
|||
HTTPS_URL + "/api/ruleChains?limit=40&textSearch=", |
|||
HttpMethod.GET, |
|||
null, |
|||
new ParameterizedTypeReference<TextPageData<RuleChain>>() { |
|||
}); |
|||
|
|||
Optional<RuleChain> 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, ExecutionException { |
|||
MqttClientConfig clientConfig = new MqttClientConfig(); |
|||
clientConfig.setClientId("MQTT client from test"); |
|||
clientConfig.setUsername(deviceCredentials.getCredentialsId()); |
|||
MqttClient mqttClient = MqttClient.create(clientConfig, listener); |
|||
mqttClient.connect("localhost", 1883).get(); |
|||
return mqttClient; |
|||
} |
|||
|
|||
@Data |
|||
private class MqttMessageListener implements MqttHandler { |
|||
private final BlockingQueue<MqttEvent> events; |
|||
|
|||
private MqttMessageListener() { |
|||
events = new ArrayBlockingQueue<>(100); |
|||
} |
|||
|
|||
@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))); |
|||
} |
|||
} |
|||
|
|||
@Data |
|||
private class MqttEvent { |
|||
private final String topic; |
|||
private final String message; |
|||
} |
|||
} |
|||
@ -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<String, Object> client; |
|||
private Map<String, Object> shared; |
|||
} |
|||
@ -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.mapper; |
|||
|
|||
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<String, List<List<Object>>> data; |
|||
private Map<String, Object> latestValues; |
|||
|
|||
public List<Object> getDataValuesByKey(String key) { |
|||
return data.entrySet().stream() |
|||
.filter(e -> e.getKey().equals(key)) |
|||
.flatMap(e -> e.getValue().stream().flatMap(Collection::stream)) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
} |
|||
@ -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 |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue