MQTT-Integrationstests in Java mit JUnit 5

<dependencies>
  <dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
  </dependency>
  <dependency>
    <groupId>com.hivemq</groupId>
    <artifactId>hivemq-mqtt-client</artifactId>
    <version>1.3.3</version>
  </dependency>
  <dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <version>4.2.2</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>1.20.3</version>
    <scope>test</scope>
  </dependency>
</dependencies>

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;

import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.*;

class PahoRoundTripTest {

  static final String BROKER =
      System.getenv().getOrDefault("MQTT_URL", "tcp://broker.connectmq.com:1883");

  MqttClient client;
  String topic;

  @BeforeEach
  void connect() throws MqttException {
    String id = UUID.randomUUID().toString().substring(0, 8);
    topic = "testmqtt/junit/" + id + "/temperature";
    client = new MqttClient(BROKER, "junit-" + id, new MemoryPersistence());

    MqttConnectOptions options = new MqttConnectOptions();
    options.setCleanSession(true);
    options.setConnectionTimeout(10);
    String user = System.getenv("MQTT_USER");
    if (user != null) {
      options.setUserName(user);
      options.setPassword(System.getenv("MQTT_PASS").toCharArray());
    }
    client.connect(options);
  }

  @AfterEach
  void disconnect() throws MqttException {
    if (client.isConnected()) client.disconnect();
    client.close();
  }

  @Test
  void publishedMessageIsReceived() throws MqttException {
    var received = new ConcurrentLinkedQueue<String>();
    client.subscribe(topic, 1, (t, msg) -> received.add(new String(msg.getPayload(), UTF_8)));

    client.publish(topic, "21.5".getBytes(UTF_8), 1, false);

    await().atMost(Duration.ofSeconds(5))
        .untilAsserted(() -> assertEquals("21.5", received.peek()));
  }
}

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.*;

import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.*;

class HiveMqRoundTripTest {

  Mqtt5BlockingClient client;
  String topic;

  @BeforeEach
  void connect() {
    String id = UUID.randomUUID().toString().substring(0, 8);
    topic = "testmqtt/junit/" + id + "/status";
    client = Mqtt5Client.builder()
        .identifier("junit-" + id)
        .serverHost("broker.connectmq.com")
        .serverPort(1883)
        .buildBlocking();
    client.connect();
  }

  @AfterEach
  void disconnect() {
    client.disconnect();
  }

  @Test
  void statusMessageIsDelivered() throws InterruptedException {
    try (var publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
      client.subscribeWith().topicFilter(topic).qos(MqttQos.AT_LEAST_ONCE).send();
      client.publishWith().topic(topic).qos(MqttQos.AT_LEAST_ONCE)
          .payload("online".getBytes(UTF_8)).send();

      Optional<Mqtt5Publish> msg = publishes.receive(5, TimeUnit.SECONDS);
      assertTrue(msg.isPresent(), "no message within 5 s on " + topic);
      assertEquals("online", new String(msg.get().getPayloadAsBytes(), UTF_8));
    }
  }
}

import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

@Testcontainers
class LocalBrokerTest {

  @Container
  static GenericContainer<?> mosquitto =
      new GenericContainer<>(DockerImageName.parse("eclipse-mosquitto:2.0"))
          .withExposedPorts(1883)
          .withCommand("mosquitto", "-c", "/mosquitto-no-auth.conf")
          .waitingFor(Wait.forLogMessage(".*mosquitto version.*running.*", 1));

  static String brokerUrl() {
    return "tcp://" + mosquitto.getHost() + ":" + mosquitto.getMappedPort(1883);
  }

  @Test
  void brokerAcceptsConnections() throws Exception {
    var client = new org.eclipse.paho.client.mqttv3.MqttClient(
        brokerUrl(), "junit-local",
        new org.eclipse.paho.client.mqttv3.persist.MemoryPersistence());
    client.connect();
    client.disconnect();
    client.close();
  }
}