|
|
@ -23,10 +23,10 @@ import io.netty.handler.codec.mqtt.MqttQoS; |
|
|
import io.netty.util.concurrent.Future; |
|
|
import io.netty.util.concurrent.Future; |
|
|
import io.netty.util.concurrent.Promise; |
|
|
import io.netty.util.concurrent.Promise; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import org.junit.After; |
|
|
import org.junit.jupiter.api.AfterEach; |
|
|
import org.junit.Assert; |
|
|
import org.junit.jupiter.api.Assertions; |
|
|
import org.junit.Before; |
|
|
import org.junit.jupiter.api.BeforeEach; |
|
|
import org.junit.Test; |
|
|
import org.junit.jupiter.api.Test; |
|
|
import org.thingsboard.common.util.AbstractListeningExecutor; |
|
|
import org.thingsboard.common.util.AbstractListeningExecutor; |
|
|
import org.thingsboard.mqtt.MqttClient; |
|
|
import org.thingsboard.mqtt.MqttClient; |
|
|
import org.thingsboard.mqtt.MqttClientConfig; |
|
|
import org.thingsboard.mqtt.MqttClientConfig; |
|
|
@ -53,7 +53,7 @@ public class MqttIntegrationTest { |
|
|
|
|
|
|
|
|
AbstractListeningExecutor handlerExecutor; |
|
|
AbstractListeningExecutor handlerExecutor; |
|
|
|
|
|
|
|
|
@Before |
|
|
@BeforeEach |
|
|
public void init() throws Exception { |
|
|
public void init() throws Exception { |
|
|
this.handlerExecutor = new AbstractListeningExecutor() { |
|
|
this.handlerExecutor = new AbstractListeningExecutor() { |
|
|
@Override |
|
|
@Override |
|
|
@ -69,7 +69,7 @@ public class MqttIntegrationTest { |
|
|
this.mqttServer.init(); |
|
|
this.mqttServer.init(); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@After |
|
|
@AfterEach |
|
|
public void destroy() throws InterruptedException { |
|
|
public void destroy() throws InterruptedException { |
|
|
if (this.mqttClient != null) { |
|
|
if (this.mqttClient != null) { |
|
|
this.mqttClient.disconnect(); |
|
|
this.mqttClient.disconnect(); |
|
|
@ -100,7 +100,7 @@ public class MqttIntegrationTest { |
|
|
|
|
|
|
|
|
log.warn("Waiting for messages acknowledgments..."); |
|
|
log.warn("Waiting for messages acknowledgments..."); |
|
|
boolean awaitResult = latch.await(10, TimeUnit.SECONDS); |
|
|
boolean awaitResult = latch.await(10, TimeUnit.SECONDS); |
|
|
Assert.assertTrue(awaitResult); |
|
|
Assertions.assertTrue(awaitResult); |
|
|
log.warn("Messages are delivered successfully..."); |
|
|
log.warn("Messages are delivered successfully..."); |
|
|
|
|
|
|
|
|
//when
|
|
|
//when
|
|
|
@ -111,7 +111,7 @@ public class MqttIntegrationTest { |
|
|
List<MqttMessageType> allReceivedEvents = this.mqttServer.getEventsFromClient(); |
|
|
List<MqttMessageType> allReceivedEvents = this.mqttServer.getEventsFromClient(); |
|
|
long disconnectCount = allReceivedEvents.stream().filter(type -> type == MqttMessageType.DISCONNECT).count(); |
|
|
long disconnectCount = allReceivedEvents.stream().filter(type -> type == MqttMessageType.DISCONNECT).count(); |
|
|
|
|
|
|
|
|
Assert.assertEquals(1, disconnectCount); |
|
|
Assertions.assertEquals(1, disconnectCount); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private Future<Void> publishMsg() { |
|
|
private Future<Void> publishMsg() { |
|
|
|