* Improve exception handling of the embedded MQTT broker so the port can be reconfigured when it is already bound and it properly unlocks files
* Rework MQTT integration tests so they each run the embedded broker on their own reserved port
Signed-off-by: Wouter Born <github@maindrain.net>
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>reserve-network-port</id>
+ <goals>
+ <goal>reserve-network-port</goal>
+ </goals>
+ <phase>process-resources</phase>
+ <configuration>
+ <portNames>
+ <portName>mqttembeddedbroker.port</portName>
+ </portNames>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
* </pre>
*/
public static final String CLIENTID = "embedded-mqtt-broker";
+
+ /**
+ * The broker persistent identifier used for identifying configurations.
+ */
+ public static final String PID = "org.openhab.core.mqttembeddedbroker";
+
+ /**
+ * The configuration key used for configuring the embedded broker port.
+ */
+ public static final String PORT = "port";
}
*
* @author David Graeff - Initial contribution
*/
-@Component(immediate = true, service = EmbeddedBrokerService.class, configurationPid = "org.openhab.core.mqttembeddedbroker", //
- property = org.osgi.framework.Constants.SERVICE_PID + "=org.openhab.core.mqttembeddedbroker")
+@Component(immediate = true, service = EmbeddedBrokerService.class, configurationPid = Constants.PID, //
+ property = org.osgi.framework.Constants.SERVICE_PID + "=" + Constants.PID)
@ConfigurableService(category = "MQTT", label = "MQTT Embedded Broker", description_uri = "mqtt:mqttembeddedbroker")
@NonNullByDefault
public class EmbeddedBrokerService
// retry starting broker, if it fails again, don't catch exception
server.startServer(new MemoryConfig(properties), null, sslContextCreator, authentificator, authorizer);
}
+ } catch (Exception e) {
+ logger.warn("Failed to start embedded MQTT server: {}", e.getMessage());
+ server.stopServer();
+ return;
}
+
this.server = server;
server.addInterceptHandler(metrics);
ScheduledExecutorService s = new ScheduledThreadPoolExecutor(1);
-runblacklist: \
bnd.identity;id='org.openhab.core.storage.json'
+-runvm: \
+ -Dio.netty.noUnsafe=true,\
+ -Dmqttembeddedbroker.port=${mqttembeddedbroker.port}
+
#
# done
#
org.openhab.core.transform;version='[3.0.0,3.0.1)',\
org.openhab.io.mqttembeddedbroker;version='[3.0.0,3.0.1)',\
org.opentest4j;version='[1.2.0,1.2.1)',\
- org.reactivestreams.reactive-streams;version='[1.0.2,1.0.3)',\
- moquette-broker;version='[0.13.0,0.13.1)'
-
--runvm: -Dio.netty.noUnsafe=true
+ org.reactivestreams.reactive-streams;version='[1.0.2,1.0.3)'
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>reserve-network-port</id>
+ <goals>
+ <goal>reserve-network-port</goal>
+ </goals>
+ <phase>process-resources</phase>
+ <configuration>
+ <portNames>
+ <portName>mqttembeddedbroker.port</portName>
+ </portNames>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.openhab.core.io.transport.mqtt.MqttService;
import org.openhab.core.io.transport.mqtt.MqttServiceObserver;
import org.openhab.io.mqttembeddedbroker.Constants;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
/**
* A full implementation test, that starts the embedded MQTT broker and publishes a homeassistant MQTT discovery device
* tree.
*
* @author David Graeff - Initial contribution
+ * @author Wouter Born - Support running MQTT itests in parallel by reconfiguring embedded broker port
*/
@NonNullByDefault
public class EmbeddedBrokerTools {
- public @Nullable MqttBrokerConnection embeddedConnection = null;
+
+ private static final int BROKER_PORT = Integer.getInteger("mqttembeddedbroker.port", 1883);
+
+ private final ConfigurationAdmin configurationAdmin;
+ private final MqttService mqttService;
+
+ public @Nullable MqttBrokerConnection embeddedConnection;
+
+ public EmbeddedBrokerTools(ConfigurationAdmin configurationAdmin, MqttService mqttService) {
+ this.configurationAdmin = configurationAdmin;
+ this.mqttService = mqttService;
+ }
/**
* Request the embedded broker connection from the {@link MqttService} and wait for a connection to be established.
*
* @throws InterruptedException
+ * @throws IOException
*/
- public MqttBrokerConnection waitForConnection(MqttService mqttService) throws InterruptedException {
+ public MqttBrokerConnection waitForConnection() throws InterruptedException, IOException {
+ reconfigurePort();
+
embeddedConnection = mqttService.getBrokerConnection(Constants.CLIENTID);
if (embeddedConnection == null) {
Semaphore semaphore = new Semaphore(1);
}
};
mqttService.addBrokersListener(observer);
- assertTrue(semaphore.tryAcquire(1000, TimeUnit.MILLISECONDS), "Wait for embedded connection client failed");
+ assertTrue(semaphore.tryAcquire(5, TimeUnit.SECONDS), "Wait for embedded connection client failed");
}
MqttBrokerConnection embeddedConnection = this.embeddedConnection;
if (embeddedConnection == null) {
if (embeddedConnection.connectionState() == MqttConnectionState.CONNECTED) {
semaphore.release();
}
- assertTrue(semaphore.tryAcquire(500, TimeUnit.MILLISECONDS), "Connection " + embeddedConnection.getClientId()
+ assertTrue(semaphore.tryAcquire(5, TimeUnit.SECONDS), "Connection " + embeddedConnection.getClientId()
+ " failed. State: " + embeddedConnection.connectionState());
return embeddedConnection;
}
+
+ public void reconfigurePort() throws IOException {
+ Configuration configuration = configurationAdmin.getConfiguration(Constants.PID, null);
+
+ Dictionary<String, Object> properties = configuration.getProperties();
+ if (properties == null) {
+ properties = new Hashtable<>();
+ }
+
+ Integer currentPort = (Integer) properties.get(Constants.PORT);
+ if (currentPort == null || currentPort.intValue() != BROKER_PORT) {
+ properties.put(Constants.PORT, BROKER_PORT);
+ configuration.update(properties);
+ // Remove the connection to make sure the test waits for the new connection to become available
+ mqttService.removeBrokerConnection(Constants.CLIENTID);
+ }
+ }
}
import org.openhab.core.types.State;
import org.openhab.core.types.UnDefType;
import org.openhab.core.util.UIDUtils;
+import org.osgi.service.cm.ConfigurationAdmin;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
*/
@NonNullByDefault
public class HomeAssistantMQTTImplementationTest extends JavaOSGiTest {
+ private @NonNullByDefault({}) ConfigurationAdmin configurationAdmin;
private @NonNullByDefault({}) MqttService mqttService;
private @NonNullByDefault({}) MqttBrokerConnection embeddedConnection;
private @NonNullByDefault({}) MqttBrokerConnection connection;
public void beforeEach() throws Exception {
registerVolatileStorageService();
mocksCloseable = openMocks(this);
+ configurationAdmin = getService(ConfigurationAdmin.class);
mqttService = getService(MqttService.class);
// Wait for the EmbeddedBrokerService internal connection to be connected
- embeddedConnection = new EmbeddedBrokerTools().waitForConnection(mqttService);
+ embeddedConnection = new EmbeddedBrokerTools(configurationAdmin, mqttService).waitForConnection();
connection = new MqttBrokerConnection(embeddedConnection.getHost(), embeddedConnection.getPort(),
embeddedConnection.isSecure(), "ha_mqtt");
- connection.start().get(1000, TimeUnit.MILLISECONDS);
+ connection.start().get(2, TimeUnit.SECONDS);
assertThat(connection.connectionState(), is(MqttConnectionState.CONNECTED));
// If the connection state changes in between -> fail
futures.add(embeddedConnection.publish(testObjectTopic + "/state", "ON".getBytes(), 0, true));
registeredTopics = futures.size();
- CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(1000, TimeUnit.MILLISECONDS);
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(2, TimeUnit.SECONDS);
failure = null;
public void afterEach() throws Exception {
if (connection != null) {
connection.removeConnectionObserver(failIfChange);
- connection.stop().get(1000, TimeUnit.MILLISECONDS);
+ connection.stop().get(2, TimeUnit.SECONDS);
}
mocksCloseable.close();
@Test
public void reconnectTest() throws InterruptedException, ExecutionException, TimeoutException {
connection.removeConnectionObserver(failIfChange);
- connection.stop().get(2000, TimeUnit.MILLISECONDS);
+ connection.stop().get(2, TimeUnit.SECONDS);
connection = new MqttBrokerConnection(embeddedConnection.getHost(), embeddedConnection.getPort(),
embeddedConnection.isSecure(), "ha_mqtt");
- connection.start().get(2000, TimeUnit.MILLISECONDS);
+ connection.start().get(2, TimeUnit.SECONDS);
}
@Test
public void retrieveAllTopics() throws InterruptedException, ExecutionException, TimeoutException {
CountDownLatch c = new CountDownLatch(registeredTopics);
connection.subscribe("homeassistant/+/+/" + ThingChannelConstants.testHomeAssistantThing.getId() + "/#",
- (topic, payload) -> c.countDown()).get(1000, TimeUnit.MILLISECONDS);
- assertTrue(c.await(1000, TimeUnit.MILLISECONDS),
+ (topic, payload) -> c.countDown()).get(2, TimeUnit.SECONDS);
+ assertTrue(c.await(2, TimeUnit.SECONDS),
"Connection " + connection.getClientId() + " not retrieving all topics");
}
return null;
});
- assertTrue(latch.await(4000, TimeUnit.MILLISECONDS));
- future.get(2000, TimeUnit.MILLISECONDS);
+ assertTrue(latch.await(4, TimeUnit.SECONDS));
+ future.get(2, TimeUnit.SECONDS);
// No failure expected and one discovered result
assertNull(failure);
-runblacklist: \
bnd.identity;id='org.openhab.core.storage.json'
+-runvm: \
+ -Dio.netty.noUnsafe=true,\
+ -Dmqttembeddedbroker.port=${mqttembeddedbroker.port}
+
#
# done
#
org.opentest4j;version='[1.2.0,1.2.1)',\
org.reactivestreams.reactive-streams;version='[1.0.2,1.0.3)',\
moquette-broker;version='[0.13.0,0.13.1)'
--runvm: -Dio.netty.noUnsafe=true
+
<groupId>com.h2database</groupId>
<artifactId>h2-mvstore</artifactId>
<version>1.4.199</version>
-
</dependency>
<dependency>
<groupId>io.netty</groupId>
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>reserve-network-port</id>
+ <goals>
+ <goal>reserve-network-port</goal>
+ </goals>
+ <phase>process-resources</phase>
+ <configuration>
+ <portNames>
+ <portName>mqttembeddedbroker.port</portName>
+ </portNames>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.openhab.core.io.transport.mqtt.MqttService;
import org.openhab.core.io.transport.mqtt.MqttServiceObserver;
import org.openhab.io.mqttembeddedbroker.Constants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
/**
* A full implementation test, that starts the embedded MQTT broker and publishes a homeassistant MQTT discovery device
* tree.
*
* @author David Graeff - Initial contribution
+ * @author Wouter Born - Support running MQTT itests in parallel by reconfiguring embedded broker port
*/
@NonNullByDefault
public class EmbeddedBrokerTools {
- private final Logger logger = LoggerFactory.getLogger(EmbeddedBrokerTools.class);
- public @Nullable MqttBrokerConnection embeddedConnection = null;
+
+ private static final int BROKER_PORT = Integer.getInteger("mqttembeddedbroker.port", 1883);
+
+ private final ConfigurationAdmin configurationAdmin;
+ private final MqttService mqttService;
+
+ public @Nullable MqttBrokerConnection embeddedConnection;
+
+ public EmbeddedBrokerTools(ConfigurationAdmin configurationAdmin, MqttService mqttService) {
+ this.configurationAdmin = configurationAdmin;
+ this.mqttService = mqttService;
+ }
/**
* Request the embedded broker connection from the {@link MqttService} and wait for a connection to be established.
*
* @throws InterruptedException
+ * @throws IOException
*/
- public MqttBrokerConnection waitForConnection(MqttService mqttService) throws InterruptedException {
+ public MqttBrokerConnection waitForConnection() throws InterruptedException, IOException {
+ reconfigurePort();
+
embeddedConnection = mqttService.getBrokerConnection(Constants.CLIENTID);
if (embeddedConnection == null) {
Semaphore semaphore = new Semaphore(1);
}
};
mqttService.addBrokersListener(observer);
- assertTrue(semaphore.tryAcquire(700, TimeUnit.MILLISECONDS), "Wait for embedded connection client failed");
+ assertTrue(semaphore.tryAcquire(5, TimeUnit.SECONDS), "Wait for embedded connection client failed");
}
MqttBrokerConnection embeddedConnection = this.embeddedConnection;
if (embeddedConnection == null) {
throw new IllegalStateException();
}
- logger.warn("waitForConnection {}", embeddedConnection.connectionState());
Semaphore semaphore = new Semaphore(1);
semaphore.acquire();
MqttConnectionObserver mqttConnectionObserver = (state, error) -> {
if (embeddedConnection.connectionState() == MqttConnectionState.CONNECTED) {
semaphore.release();
}
- assertTrue(semaphore.tryAcquire(500, TimeUnit.MILLISECONDS), "Connection " + embeddedConnection.getClientId()
+ assertTrue(semaphore.tryAcquire(5, TimeUnit.SECONDS), "Connection " + embeddedConnection.getClientId()
+ " failed. State: " + embeddedConnection.connectionState());
return embeddedConnection;
}
+
+ public void reconfigurePort() throws IOException {
+ Configuration configuration = configurationAdmin.getConfiguration(Constants.PID, null);
+
+ Dictionary<String, Object> properties = configuration.getProperties();
+ if (properties == null) {
+ properties = new Hashtable<>();
+ }
+
+ Integer currentPort = (Integer) properties.get(Constants.PORT);
+ if (currentPort == null || currentPort.intValue() != BROKER_PORT) {
+ properties.put(Constants.PORT, BROKER_PORT);
+ configuration.update(properties);
+ // Remove the connection to make sure the test waits for the new connection to become available
+ mqttService.removeBrokerConnection(Constants.CLIENTID);
+ }
+ }
}
import org.openhab.core.library.types.OnOffType;
import org.openhab.core.test.java.JavaOSGiTest;
import org.openhab.core.types.UnDefType;
+import org.osgi.service.cm.ConfigurationAdmin;
/**
* A full implementation test, that starts the embedded MQTT broker and publishes a homie device tree.
private static final String DEVICE_ID = ThingChannelConstants.testHomieThing.getId();
private static final String DEVICE_TOPIC = BASE_TOPIC + "/" + DEVICE_ID;
+ private @NonNullByDefault({}) ConfigurationAdmin configurationAdmin;
private @NonNullByDefault({}) MqttService mqttService;
private @NonNullByDefault({}) MqttBrokerConnection embeddedConnection;
private @NonNullByDefault({}) MqttBrokerConnection connection;
public void beforeEach() throws Exception {
registerVolatileStorageService();
mocksCloseable = openMocks(this);
+ configurationAdmin = getService(ConfigurationAdmin.class);
mqttService = getService(MqttService.class);
- embeddedConnection = new EmbeddedBrokerTools().waitForConnection(mqttService);
+ // Wait for the EmbeddedBrokerService internal connection to be connected
+ embeddedConnection = new EmbeddedBrokerTools(configurationAdmin, mqttService).waitForConnection();
embeddedConnection.setQos(1);
connection = new MqttBrokerConnection(embeddedConnection.getHost(), embeddedConnection.getPort(),
embeddedConnection.isSecure(), "homie");
connection.setQos(1);
- connection.start().get(500, TimeUnit.MILLISECONDS);
+ connection.start().get(5, TimeUnit.SECONDS);
assertThat(connection.connectionState(), is(MqttConnectionState.CONNECTED));
// If the connection state changes in between -> fail
connection.addConnectionObserver(failIfChange);
futures.add(publish(propertyTestTopic + "/$datatype", "boolean"));
registeredTopics = futures.size();
- CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(1000, TimeUnit.MILLISECONDS);
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(2, TimeUnit.SECONDS);
scheduler = new ScheduledThreadPoolExecutor(6);
}
public void afterEach() throws Exception {
if (connection != null) {
connection.removeConnectionObserver(failIfChange);
- connection.stop().get(500, TimeUnit.MILLISECONDS);
+ connection.stop().get(2, TimeUnit.SECONDS);
+ }
+ if (scheduler != null) {
+ scheduler.shutdownNow();
}
- scheduler.shutdownNow();
mocksCloseable.close();
}
public void retrieveAllTopics() throws InterruptedException, ExecutionException, TimeoutException {
// four topics are not under /testnode !
CountDownLatch c = new CountDownLatch(registeredTopics - 4);
- connection.subscribe(DEVICE_TOPIC + "/testnode/#", (topic, payload) -> c.countDown()).get(5000,
- TimeUnit.MILLISECONDS);
- assertTrue(c.await(5000, TimeUnit.MILLISECONDS),
+ connection.subscribe(DEVICE_TOPIC + "/testnode/#", (topic, payload) -> c.countDown()).get(5, TimeUnit.SECONDS);
+ assertTrue(c.await(5, TimeUnit.SECONDS),
"Connection " + connection.getClientId() + " not retrieving all topics ");
}
-runblacklist: \
bnd.identity;id='org.openhab.core.storage.json'
+-runvm: \
+ -Dio.netty.noUnsafe=true,\
+ -Dmqttembeddedbroker.port=${mqttembeddedbroker.port}
+
#
# done
#
org.opentest4j;version='[1.2.0,1.2.1)',\
org.reactivestreams.reactive-streams;version='[1.0.2,1.0.3)',\
moquette-broker;version='[0.13.0,0.13.1)'
--runvm: -Dio.netty.noUnsafe=true
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>reserve-network-port</id>
+ <goals>
+ <goal>reserve-network-port</goal>
+ </goals>
+ <phase>process-resources</phase>
+ <configuration>
+ <portNames>
+ <portName>mqttembeddedbroker.port</portName>
+ </portNames>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.openhab.core.io.transport.mqtt.MqttConnectionState;
import org.openhab.core.io.transport.mqtt.MqttService;
import org.openhab.core.io.transport.mqtt.MqttServiceObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
/**
* A full implementation test, that starts the embedded MQTT broker and publishes a homeassistant MQTT discovery device
* tree.
*
* @author David Graeff - Initial contribution
+ * @author Wouter Born - Support running MQTT itests in parallel by reconfiguring embedded broker port
*/
@NonNullByDefault
public class EmbeddedBrokerTools {
- private final Logger logger = LoggerFactory.getLogger(EmbeddedBrokerTools.class);
- public @Nullable MqttBrokerConnection embeddedConnection = null;
+
+ private static final int BROKER_PORT = Integer.getInteger("mqttembeddedbroker.port", 1883);
+
+ private final ConfigurationAdmin configurationAdmin;
+ private final MqttService mqttService;
+
+ public @Nullable MqttBrokerConnection embeddedConnection;
+
+ public EmbeddedBrokerTools(ConfigurationAdmin configurationAdmin, MqttService mqttService) {
+ this.configurationAdmin = configurationAdmin;
+ this.mqttService = mqttService;
+ }
/**
* Request the embedded broker connection from the {@link MqttService} and wait for a connection to be established.
*
* @throws InterruptedException
+ * @throws IOException
*/
- public MqttBrokerConnection waitForConnection(MqttService mqttService) throws InterruptedException {
+ public MqttBrokerConnection waitForConnection() throws InterruptedException, IOException {
+ reconfigurePort();
+
embeddedConnection = mqttService.getBrokerConnection(Constants.CLIENTID);
if (embeddedConnection == null) {
Semaphore semaphore = new Semaphore(1);
}
};
mqttService.addBrokersListener(observer);
- assertTrue(semaphore.tryAcquire(700, TimeUnit.MILLISECONDS), "Wait for embedded connection client failed");
+ assertTrue(semaphore.tryAcquire(5, TimeUnit.SECONDS), "Wait for embedded connection client failed");
}
MqttBrokerConnection embeddedConnection = this.embeddedConnection;
if (embeddedConnection == null) {
throw new IllegalStateException();
}
- logger.warn("waitForConnection {}", embeddedConnection.connectionState());
Semaphore semaphore = new Semaphore(1);
semaphore.acquire();
MqttConnectionObserver mqttConnectionObserver = (state, error) -> {
if (embeddedConnection.connectionState() == MqttConnectionState.CONNECTED) {
semaphore.release();
}
- assertTrue(semaphore.tryAcquire(500, TimeUnit.MILLISECONDS), "Connection " + embeddedConnection.getClientId()
+ assertTrue(semaphore.tryAcquire(5, TimeUnit.SECONDS), "Connection " + embeddedConnection.getClientId()
+ " failed. State: " + embeddedConnection.connectionState());
return embeddedConnection;
}
+
+ public void reconfigurePort() throws IOException {
+ Configuration configuration = configurationAdmin.getConfiguration(Constants.PID, null);
+
+ Dictionary<String, Object> properties = configuration.getProperties();
+ if (properties == null) {
+ properties = new Hashtable<>();
+ }
+
+ Integer currentPort = (Integer) properties.get(Constants.PORT);
+ if (currentPort == null || currentPort.intValue() != BROKER_PORT) {
+ properties.put(Constants.PORT, BROKER_PORT);
+ configuration.update(properties);
+ // Remove the connection to make sure the test waits for the new connection to become available
+ mqttService.removeBrokerConnection(Constants.CLIENTID);
+ }
+ }
}
import org.openhab.core.io.transport.mqtt.MqttConnectionState;
import org.openhab.core.io.transport.mqtt.MqttService;
import org.openhab.core.test.java.JavaOSGiTest;
+import org.osgi.service.cm.ConfigurationAdmin;
/**
* Moquette test
private @NonNullByDefault({}) AutoCloseable mocksCloseable;
+ private @NonNullByDefault({}) ConfigurationAdmin configurationAdmin;
private @NonNullByDefault({}) MqttService mqttService;
private @NonNullByDefault({}) MqttBrokerConnection embeddedConnection;
private @NonNullByDefault({}) MqttBrokerConnection clientConnection;
public void beforeEach() throws Exception {
registerVolatileStorageService();
mocksCloseable = openMocks(this);
+ configurationAdmin = getService(ConfigurationAdmin.class);
mqttService = getService(MqttService.class);
// Wait for the EmbeddedBrokerService internal connection to be connected
- embeddedConnection = new EmbeddedBrokerTools().waitForConnection(mqttService);
+ embeddedConnection = new EmbeddedBrokerTools(configurationAdmin, mqttService).waitForConnection();
embeddedConnection.setQos(1);
clientConnection = new MqttBrokerConnection(embeddedConnection.getHost(), embeddedConnection.getPort(),
projects / openhab-addons.git / commitdiff