9 changed files with 531 additions and 0 deletions
@ -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. <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: |
|||
|
|||
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 |
|||
@ -0,0 +1,98 @@ |
|||
<!-- |
|||
|
|||
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>integration-tests</artifactId> |
|||
|
|||
<name>ThingsBoard Integration Tests</name> |
|||
<url>https://thingsboard.io</url> |
|||
<description>Project for ThingsBoard integration tests with using Docker</description> |
|||
|
|||
<properties> |
|||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> |
|||
<main.dir>${basedir}/../..</main.dir> |
|||
<integrationtests.skip>true</integrationtests.skip> |
|||
<testcontainers.version>1.9.1</testcontainers.version> |
|||
<java-websocket.version>1.3.9</java-websocket.version> |
|||
</properties> |
|||
|
|||
<dependencies> |
|||
<dependency> |
|||
<groupId>org.testcontainers</groupId> |
|||
<artifactId>testcontainers</artifactId> |
|||
<version>${testcontainers.version}</version> |
|||
</dependency> |
|||
<dependency> |
|||
<groupId>org.java-websocket</groupId> |
|||
<artifactId>Java-WebSocket</artifactId> |
|||
<version>${java-websocket.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>${integrationtests.skip}</skipTests> |
|||
</configuration> |
|||
</plugin> |
|||
</plugins> |
|||
</build> |
|||
|
|||
</project> |
|||
@ -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<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; |
|||
} |
|||
|
|||
} |
|||
@ -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")); |
|||
} |
|||
@ -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<String> 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; |
|||
} |
|||
} |
|||
@ -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<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,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()); |
|||
} |
|||
} |
|||
@ -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<MqttEvent> 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; |
|||
} |
|||
} |
|||
Loading…
Reference in new issue