diff --git a/manager/src/main/java/org/openremote/manager/mqtt/MQTTBrokerService.java b/manager/src/main/java/org/openremote/manager/mqtt/MQTTBrokerService.java index a39754b697..45a2c2cca1 100644 --- a/manager/src/main/java/org/openremote/manager/mqtt/MQTTBrokerService.java +++ b/manager/src/main/java/org/openremote/manager/mqtt/MQTTBrokerService.java @@ -464,11 +464,21 @@ public void connectionFailed( connectionIDConnectionMap.remove(getConnectionIDString(connection)); if (connection.getClientID() != null) { - RemotingConnection remotingConnection = - clientIDConnectionMap.remove(connection.getClientID()); - if (remotingConnection != null) { - disconnectedConnectionCache.put(connection.getClientID(), remotingConnection); - } + // Check and remove atomically so a reconnect cannot replace this entry mid-cleanup. + // If it is already absent, leave any newer disconnected connection in the cache. + clientIDConnectionMap.computeIfPresent( + connection.getClientID(), + (clientID, currentConnection) -> { + // Only clean up the exact connection that triggered this callback. + if (currentConnection == connection) { + // Last wills are processed asynchronously and still need this connection. + // Cache it before removal so lookups never see a gap between the two stores. + disconnectedConnectionCache.put(clientID, connection); + return null; // Returning null removes the active entry. + } + // A reconnect has replaced this connection; preserve the replacement and cache. + return currentConnection; + }); } if (exception.getType() diff --git a/manager/test/src/groovy/org/openremote/manager/mqtt/MQTTBrokerServiceTest.groovy b/manager/test/src/groovy/org/openremote/manager/mqtt/MQTTBrokerServiceTest.groovy new file mode 100644 index 0000000000..a478dd3e2a --- /dev/null +++ b/manager/test/src/groovy/org/openremote/manager/mqtt/MQTTBrokerServiceTest.groovy @@ -0,0 +1,147 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.manager.mqtt + +import com.google.common.cache.CacheBuilder +import org.apache.activemq.artemis.api.core.ActiveMQException +import org.apache.activemq.artemis.api.core.ActiveMQExceptionType +import org.apache.activemq.artemis.core.remoting.FailureListener +import org.apache.activemq.artemis.core.remoting.server.RemotingService +import org.apache.activemq.artemis.core.server.ActiveMQServer +import org.apache.activemq.artemis.core.server.embedded.EmbeddedActiveMQ +import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection +import spock.lang.Specification + +import java.util.concurrent.ConcurrentHashMap +import java.util.function.BiFunction + +class MQTTBrokerServiceTest extends Specification { + + private MQTTBrokerService brokerService = new MQTTBrokerService() + private FailureListener failureListener + + def setup() { + def remotingService = Stub(RemotingService) { + getConnections() >> Collections.emptySet() + } + def server = Stub(ActiveMQServer) { + getRemotingService() >> remotingService + } + brokerService.@server = Stub(EmbeddedActiveMQ) { + getActiveMQServer() >> server + } + brokerService.@disconnectedConnectionCache = CacheBuilder.newBuilder().build() + } + + def "a last will can resolve its connection during disconnect cleanup"() { + given: "a connection whose transport has already closed" + def connection = connection("original") + def connections = new ConnectionMap() + connections.put("client", connection) + brokerService.@clientIDConnectionMap = connections + brokerService.afterCreateConnection(connection) + + and: "a publish lookup interleaved with removing the active connection" + RemotingConnection resolvedConnection + connections.afterRemove = { + resolvedConnection = brokerService.getConnectionFromClientID("client") + } + + when: "disconnect cleanup removes the connection from the active map" + failureListener.connectionFailed(new ActiveMQException(ActiveMQExceptionType.DISCONNECTED), false) + + then: "the publish can still resolve the connection for the last will" + resolvedConnection.is(connection) + brokerService.getConnectionFromClientID("client").is(connection) + connections.isEmpty() + } + + def "a delayed disconnect callback preserves a reconnected client's connection"() { + given: "a client that reconnected before its old disconnect callback runs" + def originalConnection = connection("original") + def replacementConnection = connection("replacement") + brokerService.afterCreateConnection(originalConnection) + brokerService.@clientIDConnectionMap.put("client", replacementConnection) + + when: "the original connection's disconnect callback runs" + failureListener.connectionFailed(new ActiveMQException(ActiveMQExceptionType.DISCONNECTED), false) + + then: "the replacement remains active and is not recorded as disconnected" + brokerService.@clientIDConnectionMap.get("client").is(replacementConnection) + brokerService.getConnectionFromClientID("client").is(replacementConnection) + brokerService.@disconnectedConnectionCache.getIfPresent("client") == null + } + + def "a late old disconnect does not replace the newer disconnected connection"() { + given: "a client that reconnected before its old disconnect callback runs" + def originalConnection = connection("original") + def replacementConnection = connection("replacement") + brokerService.afterCreateConnection(originalConnection) + def originalFailureListener = failureListener + brokerService.afterCreateConnection(replacementConnection) + def replacementFailureListener = failureListener + brokerService.@clientIDConnectionMap.put("client", replacementConnection) + + when: "the replacement also disconnects" + replacementFailureListener.connectionFailed(new ActiveMQException(ActiveMQExceptionType.DISCONNECTED), false) + + then: "its connection can be resolved for pending messages" + brokerService.getConnectionFromClientID("client").is(replacementConnection) + + when: "the original connection's callback arrives late" + originalFailureListener.connectionFailed(new ActiveMQException(ActiveMQExceptionType.DISCONNECTED), false) + + then: "pending messages still resolve the latest connection" + brokerService.getConnectionFromClientID("client").is(replacementConnection) + } + + private RemotingConnection connection(String id) { + Stub(RemotingConnection) { + getID() >> id + getClientID() >> "client" + addFailureListener(_) >> { FailureListener listener -> failureListener = listener } + } + } + + /** Runs a concurrent publish lookup at the point where the active entry has just been removed. */ + private static class ConnectionMap extends ConcurrentHashMap { + Runnable afterRemove + + @Override + RemotingConnection remove(Object key) { + def removed = super.remove(key) + afterRemove.run() + return removed + } + + @Override + boolean remove(Object key, Object value) { + def removed = super.remove(key, value) + afterRemove.run() + return removed + } + + @Override + RemotingConnection computeIfPresent(String key, BiFunction remappingFunction) { + def remaining = super.computeIfPresent(key, remappingFunction) + afterRemove.run() + return remaining + } + } +} diff --git a/test/src/test/groovy/org/openremote/test/mqtt/MqttBrokerTest.groovy b/test/src/test/groovy/org/openremote/test/mqtt/MqttBrokerTest.groovy index 5a682089a8..8508a4506a 100644 --- a/test/src/test/groovy/org/openremote/test/mqtt/MqttBrokerTest.groovy +++ b/test/src/test/groovy/org/openremote/test/mqtt/MqttBrokerTest.groovy @@ -18,12 +18,14 @@ */ package org.openremote.test.mqtt +import com.google.common.cache.Cache import com.google.common.cache.CacheBuilder import com.hivemq.client.internal.mqtt.mqtt3.Mqtt3AsyncClientView import com.hivemq.client.internal.mqtt.mqtt3.Mqtt3ClientConfigView import com.hivemq.client.mqtt.MqttClientConfig import com.hivemq.client.mqtt.MqttClientConnectionConfig import io.netty.channel.socket.SocketChannel +import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection import org.openremote.agent.protocol.mqtt.MQTTLastWill import org.openremote.agent.protocol.mqtt.MQTTMessage import org.openremote.agent.protocol.mqtt.MQTT_IOClient @@ -40,6 +42,7 @@ import org.openremote.model.asset.impl.RoomAsset import org.openremote.model.asset.impl.ThingAsset import org.openremote.model.attribute.Attribute import org.openremote.model.attribute.AttributeEvent +import org.openremote.model.attribute.AttributeMap import org.openremote.model.attribute.MetaItem import org.openremote.model.auth.UsernamePassword import org.openremote.model.event.shared.SharedEvent @@ -52,7 +55,10 @@ import spock.lang.Specification import spock.util.concurrent.PollingConditions import java.time.Duration +import java.util.concurrent.BlockingQueue import java.util.concurrent.CopyOnWriteArrayList +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.TimeUnit import java.util.function.Consumer import static org.openremote.model.util.MapAccess.getInteger @@ -63,34 +69,106 @@ import static org.openremote.model.value.ValueType.TEXT class MqttBrokerTest extends Specification implements ManagerContainerTrait { - def "Mqtt broker event test"() { - given: "the container environment is started" - List receivedEvents = new CopyOnWriteArrayList<>() - List restrictedReceivedEvents = new CopyOnWriteArrayList<>() - List receivedValues = new CopyOnWriteArrayList<>() - List clientConnectionStatuses = new CopyOnWriteArrayList<>() - MQTT_IOClient client = null - MQTT_IOClient newClient = null - def conditions = new PollingConditions(timeout: 15, initialDelay: 0.1, delay: 0.2) - def container = startContainer(defaultConfig(), defaultServices()) - def managerTestSetup = container.getService(SetupService.class).getTaskOfType(ManagerTestSetup.class) - def keycloakTestSetup = container.getService(SetupService.class).getTaskOfType(KeycloakTestSetup.class) - def mqttBrokerService = container.getService(MQTTBrokerService.class) - def assetStorageService = container.getService(AssetStorageService.class) - def assetProcessingService = container.getService(AssetProcessingService.class) - def defaultMQTTHandler = mqttBrokerService.getCustomHandlers().find { + private ManagerTestSetup managerTestSetup + private KeycloakTestSetup keycloakTestSetup + private MQTTBrokerService mqttBrokerService + private AssetStorageService assetStorageService + private AssetProcessingService assetProcessingService + private DefaultMQTTHandler defaultMQTTHandler + private Cache originalDisconnectedConnectionCache + private String mqttHost + private int mqttPort + private UsernamePassword mqttCredentials + private UsernamePassword restrictedMqttCredentials + private final List clients = [] + private final Map originalAttributes = [:] + private final List createdAssetIds = [] + private List originalUserAssetLinks + private final PollingConditions conditions = new PollingConditions(timeout: 15, initialDelay: 0.1, delay: 0.2) + + // MQTT callbacks add events asynchronously. Blocking queues let absence checks use timed poll() + // to wait for unexpected events, returning null only after the full observation period. + // PollingConditions.eventually { ...isEmpty() } would succeed on the first empty check. + private final BlockingQueue receivedEvents = new LinkedBlockingQueue<>() + private final Consumer> eventConsumer = eventConsumerFor(receivedEvents) + + def setupSpec() { + startContainer(defaultConfig(), defaultServices()) + } + + def setup() { + managerTestSetup = container.getService(SetupService.class).getTaskOfType(ManagerTestSetup.class) + keycloakTestSetup = container.getService(SetupService.class).getTaskOfType(KeycloakTestSetup.class) + mqttBrokerService = container.getService(MQTTBrokerService.class) + assetStorageService = container.getService(AssetStorageService.class) + assetProcessingService = container.getService(AssetProcessingService.class) + defaultMQTTHandler = mqttBrokerService.getCustomHandlers().find { it instanceof DefaultMQTTHandler } as DefaultMQTTHandler - def mqttClientId = UniqueIdentifierGenerator.generateId() - def username = keycloakTestSetup.realmBuilding.name + ":" + keycloakTestSetup.serviceUser.username // realm and OAuth client id - def password = keycloakTestSetup.serviceUser.secret + originalDisconnectedConnectionCache = mqttBrokerService.@disconnectedConnectionCache + mqttHost = getString(container.getConfig(), MQTT_SERVER_LISTEN_HOST, "0.0.0.0") + mqttPort = getInteger(container.getConfig(), MQTT_SERVER_LISTEN_PORT, 1883) + mqttCredentials = new UsernamePassword( + "${keycloakTestSetup.realmBuilding.name}:${keycloakTestSetup.serviceUser.username}".toString(), + keycloakTestSetup.serviceUser.secret) + restrictedMqttCredentials = new UsernamePassword( + "${keycloakTestSetup.realmBuilding.name}:${keycloakTestSetup.serviceUser2.username}".toString(), + keycloakTestSetup.serviceUser2.secret) + } - def mqttHost = getString(container.getConfig(), MQTT_SERVER_LISTEN_HOST, "0.0.0.0") - def mqttPort = getInteger(container.getConfig(), MQTT_SERVER_LISTEN_PORT, 1883) + def cleanup() { + try { + clients.each { it.disconnect() } + if (mqttBrokerService != null) { + conditions.eventually { + assert clients.every { it.getConnectionStatus() == ConnectionStatus.DISCONNECTED } + assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id).isEmpty() + assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).isEmpty() + assert defaultMQTTHandler.sessionSubscriptionConsumers.isEmpty() + } + } + + // Restore fixtures only after clients and subscriptions are gone, so cleanup cannot trigger reconnects + // or deliver restoration events to a test client. + if (!createdAssetIds.isEmpty()) { + assert assetStorageService.deleteUntilFinished(createdAssetIds).get(15, TimeUnit.SECONDS) + } + originalAttributes.each { assetId, attributes -> + def asset = assetStorageService.find(assetId) + asset.setAttributes(attributes) + assetStorageService.merge(asset) + } + if (originalUserAssetLinks != null) { + def currentLinks = assetStorageService.findUserAssetLinks(keycloakTestSetup.realmBuilding.name, + keycloakTestSetup.serviceUser2.id, null as String) + assetStorageService.deleteUserAssetLinks(currentLinks.findAll { current -> + !originalUserAssetLinks.any { it.id == current.id } + }) + assetStorageService.storeUserAssetLinks(originalUserAssetLinks.findAll { original -> + !currentLinks.any { it.id == original.id } + }) + } + if (!originalAttributes.isEmpty()) { + conditions.eventually { + assert noEventProcessedIn(assetProcessingService, 300) + } + } + } finally { + if (originalDisconnectedConnectionCache != null) { + mqttBrokerService.@disconnectedConnectionCache = originalDisconnectedConnectionCache + } + } + } + + def "MQTT connections reject invalid credentials and release resources on disconnect"() { + given: "MQTT client credentials and a connection status listener" + def mqttClientId = UniqueIdentifierGenerator.generateId() + List clientConnectionStatuses = new CopyOnWriteArrayList<>() + MQTT_IOClient client when: "a mqtt client connects with invalid credentials" def wrongUsername = "master:" + keycloakTestSetup.serviceUser.username - client = new MQTT_IOClient(mqttClientId, mqttHost, mqttPort, false, true, new UsernamePassword(wrongUsername, password), null, null) + client = createMqttClient(mqttClientId, new UsernamePassword(wrongUsername, mqttCredentials.password)) client.addConnectionStatusConsumer {clientConnectionStatuses.add(it)} client.connect() @@ -110,7 +188,7 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { when: "a mqtt client connects with valid credentials" mqttClientId = UniqueIdentifierGenerator.generateId() - client = new MQTT_IOClient(mqttClientId, mqttHost, mqttPort, false, true, new UsernamePassword(username, password), null, null) + client = createMqttClient(mqttClientId) client.connect() then: "mqtt connection should exist" @@ -120,7 +198,6 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { } when: "the client disconnects" - mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] client.disconnect() then: "the client resources should be freed" @@ -128,16 +205,16 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { assert client.getConnectionStatus() == ConnectionStatus.DISCONNECTED assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id).isEmpty() } + } - when: "a mqtt client connects with valid credentials" + def "MQTT subscriptions reject another realm and missing or mismatched client IDs"() { + given: "a connected client that removes consumers after subscription failures" + def mqttClientId = UniqueIdentifierGenerator.generateId() List subFailures = new CopyOnWriteArrayList<>() - mqttClientId = UniqueIdentifierGenerator.generateId() - client = new MQTT_IOClient(mqttClientId, mqttHost, mqttPort, false, true, new UsernamePassword(username, password), null, null) - client.setTopicSubscribeFailureConsumer {subFailures.add(it)} + def client = createMqttClient(mqttClientId) + client.setTopicSubscribeFailureConsumer { subFailures.add(it) } client.setRemoveConsumersOnSubscriptionFailure(true) client.connect() - - then: "mqtt connection should exist" conditions.eventually { assert client.getConnectionStatus() == ConnectionStatus.CONNECTED assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id).size() == 1 @@ -147,26 +224,20 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { def topic = "${keycloakTestSetup.realmCity.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.thingId".toString() client.addMessageConsumer(topic, {msg -> }) - then: "No subscription should exist" + then: "no subscription should exist" conditions.eventually { - // A client re-subscribes its retained consumers whenever it reconnects, so failures can be reported more - // than once and the total count is not stable - assert subFailures.contains(topic) - assert client.topicConsumerMap.get(topic) == null // Consumer added and removed on failure - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertSubscriptionFailed(client, topic, subFailures) + assertNoBrokerSubscriptions(keycloakTestSetup.serviceUser.id) } when: "a mqtt client subscribes with clientId missing" topic = "${keycloakTestSetup.realmBuilding.name}/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() client.addMessageConsumer(topic, {msg ->}) - then: "No subscription should exist" + then: "no subscription should exist" conditions.eventually { - assert subFailures.contains(topic) - assert client.topicConsumerMap.get(topic) == null // Consumer added and removed on failure - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertSubscriptionFailed(client, topic, subFailures) + assertNoBrokerSubscriptions(keycloakTestSetup.serviceUser.id) } when: "a mqtt client subscribes with different clientId" @@ -174,58 +245,45 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() client.addMessageConsumer(topic, {msg ->}) - then: "No subscription should exist" + then: "no subscription should exist" conditions.eventually { - assert subFailures.contains(topic) - assert client.topicConsumerMap.get(topic) == null // Consumer added and removed on failure - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id).size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertSubscriptionFailed(client, topic, subFailures) + assertNoBrokerSubscriptions(keycloakTestSetup.serviceUser.id) } + } + + def "MQTT attribute subscriptions receive events and writes and stop after unsubscribe"() { + given: "a connected MQTT client" + preserveAttributes(managerTestSetup.apartment1HallwayId) + def mqttClientId = UniqueIdentifierGenerator.generateId() + def client = connectMqttClient(mqttClientId) when: "a mqtt client subscribes to all attributes of an asset" - topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() - Consumer> eventConsumer = { msg -> - def event = ValueUtil.parse(msg.payload as String, SharedEvent.class) - receivedEvents.add(event.get()) - } + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() client.addMessageConsumer(topic, eventConsumer) - then: "A subscription should exist" + then: "a subscription should exist" conditions.eventually { - assert client.topicConsumerMap.get(topic) != null - assert client.topicConsumerMap.get(topic).consumers.size() == 1 - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id).size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 1 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser.id, 1) } when: "An attribute event occurs for a subscribed attribute" def attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "motionSensor", 50) assetProcessingService.sendAttributeEvent(attributeEvent) - then: "A publish event message should be sent" + then: "a publish event message should be sent" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 50 + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 50) } receivedEvents.clear() - when: "Another asset attribute changed the client is subscribed on" + when: "another asset attribute changed the client is subscribed on" attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "presenceDetected", true) assetProcessingService.sendAttributeEvent(attributeEvent) - then: "A second publish event message should be sent" + then: "a second publish event message should be sent" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "presenceDetected" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(false) == true + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "presenceDetected", true) } receivedEvents.clear() @@ -234,14 +292,10 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { def payload = "70" client.sendMessage(new MQTTMessage(topic, payload)) - then: "The value of the attribute should be updated and the client should have received the event" + then: "the value of the attribute should be updated and the client should have received the event" new PollingConditions(initialDelay: 1, timeout: 10, delay: 1).eventually { assert assetStorageService.find(managerTestSetup.apartment1HallwayId).getAttribute("motionSensor").get().value.orElse(0) == 70d - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 70d + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 70d) } receivedEvents.clear() @@ -253,11 +307,7 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the value of the attribute should be updated and the client should have received the event" conditions.eventually { assert !assetStorageService.find(managerTestSetup.apartment1HallwayId).getAttribute("lights").get().value.orElse(true) - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "lights" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(true) == false + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "lights", false) } receivedEvents.clear() @@ -269,11 +319,7 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the value of the attribute should be updated and the client should have received the event" conditions.eventually { assert assetStorageService.find(managerTestSetup.apartment1HallwayId).getAttribute("lights").get().value.orElse(false) - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "lights" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(false) == true + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "lights", true) } receivedEvents.clear() @@ -281,21 +327,17 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() client.removeMessageConsumer(topic, eventConsumer) - then: "No subscription should exist" + then: "no subscription should exist" conditions.eventually { - assert client.topicConsumerMap.get(topic) == null - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertUnsubscribed(client, topic, keycloakTestSetup.serviceUser.id) } when: "another asset attribute changed without any subscriptions" attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "presenceDetected", false) assetProcessingService.sendAttributeEvent(attributeEvent) - then: "No publish event message should be sent" - new PollingConditions(initialDelay: 1, delay: 1, timeout: 10).eventually { - assert receivedEvents.size() == 0 - } + then: "no publish event message should be sent" + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null when: "a mqtt client subscribes to a specific asset attribute" topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/motionSensor/$managerTestSetup.apartment1HallwayId".toString() @@ -303,11 +345,7 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the subscription should be created" conditions.eventually { - assert client.topicConsumerMap.get(topic) != null - assert client.topicConsumerMap.get(topic).consumers.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 1 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser.id, 1) } when: "that attribute changes" @@ -316,11 +354,7 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "a publish event message should be sent to the client" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 30 + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 30) } receivedEvents.clear() @@ -329,44 +363,40 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { assetProcessingService.sendAttributeEvent(attributeEvent) then: "no publish event message should be sent" - new PollingConditions(initialDelay: 1, delay: 1, timeout: 10).eventually { - assert receivedEvents.size() == 0 - } + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null when: "a mqtt client unsubscribes from an asset attribute" client.removeMessageConsumer(topic, eventConsumer) - then: "No subscription should exist" + then: "no subscription should exist" conditions.eventually { - assert client.topicConsumerMap.get(topic) == null - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertUnsubscribed(client, topic, keycloakTestSetup.serviceUser.id) } + } + + def "MQTT wildcard and value subscriptions filter events and survive reconnects"() { + given: "a connected MQTT client and a collector for raw attribute values" + preserveAttributes(managerTestSetup.apartment1HallwayId) + def mqttClientId = UniqueIdentifierGenerator.generateId() + def client = connectMqttClient(mqttClientId) + List receivedValues = new CopyOnWriteArrayList<>() when: "a mqtt client subscribes to attributes for descendants of an asset" - topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() client.addMessageConsumer(topic, eventConsumer) then: "a subscription should exist" conditions.eventually { - assert client.topicConsumerMap.get(topic) != null - assert client.topicConsumerMap.get(topic).consumers.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 1 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser.id, 1) } when: "a child asset of the subscription attribute event occurs" - attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "motionSensor", 40) + def attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "motionSensor", 40) assetProcessingService.sendAttributeEvent(attributeEvent) then: "a publish event message should be sent" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 40 + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 40) } receivedEvents.clear() @@ -379,41 +409,29 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the subscription should be created" conditions.eventually { - assert client.topicConsumerMap.get(topic) != null - assert client.topicConsumerMap.get(topic).consumers.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 2 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser.id, 2) } when: "a subscribed attribute changes" attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "motionSensor", 50) assetProcessingService.sendAttributeEvent(attributeEvent) - then: "A publish value message should be sent to both subscriptions" + then: "a publish value message should be sent to both subscriptions" conditions.eventually { assert receivedValues.size() == 1 assert receivedValues.get(0) == "50" - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 50 + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 50) } receivedEvents.clear() receivedValues.clear() when: "a subscribed attribute changes" - attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "presenceDetected", false) + attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "presenceDetected", true) assetProcessingService.sendAttributeEvent(attributeEvent) then: "only the wildcard subscription should have received a publish" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "presenceDetected" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(true) == false + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "presenceDetected", true) assert receivedValues.size() == 0 } @@ -422,11 +440,7 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "only one subscription should remain" conditions.eventually { - assert client.topicConsumerMap.get(topic) == null - assert client.topicConsumerMap.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 1 + assertUnsubscribed(client, topic, keycloakTestSetup.serviceUser.id, 1) } when: "a client disconnects" @@ -446,20 +460,25 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { conditions.eventually { assert client.getConnectionStatus() == ConnectionStatus.CONNECTED assert client.topicConsumerMap.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 1 + client.topicConsumerMap.keySet().each { subscriptionTopic -> + assertSubscribed(client, subscriptionTopic, keycloakTestSetup.serviceUser.id, 1) + } } + } + + def "MQTT asset subscriptions receive realm children and descendants and stop after unsubscribe"() { + given: "a connected MQTT client" + preserveAttributes(managerTestSetup.smartBuildingId) + def mqttClientId = UniqueIdentifierGenerator.generateId() + def client = connectMqttClient(mqttClientId) when: "a mqtt client subscribes to assets that are direct children of the realm" - topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ASSET_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD".toString() + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ASSET_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD".toString() client.addMessageConsumer(topic, eventConsumer) - then: "A subscription should exist" + then: "a subscription should exist" conditions.eventually { - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 2 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser.id, 1) } when: "an asset is updated with a new attribute" @@ -467,15 +486,11 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { asset1.addAttributes(new Attribute<>("temp", TEXT, "hello world")) assetStorageService.merge(asset1) - then: "A publish event message should be sent" + then: "a publish event message should be sent" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AssetEvent - assert (receivedEvents.get(0) as AssetEvent).id == managerTestSetup.smartBuildingId - assert (receivedEvents.get(0) as AssetEvent).asset.attributes.get("temp") != null - assert (receivedEvents.get(0) as AssetEvent).asset.attributes.get("temp").flatMap(){ - it.getValue() - }.orElse(null) == "hello world" + def event = assertSingleEvent(receivedEvents, AssetEvent) + assert event.id == managerTestSetup.smartBuildingId + assert event.asset.getAttribute("temp").flatMap { it.value }.orElse(null) == "hello world" } receivedEvents.clear() @@ -485,186 +500,187 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the subscription should exist" conditions.eventually { - assert client.topicConsumerMap.get(topic) != null - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 3 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser.id, 2) } when: "an asset is added as a descendant to the subscribed asset" def childAsset = new ThingAsset("child") + .setId(UniqueIdentifierGenerator.generateId()) .setParentId(managerTestSetup.apartment1LivingroomId) .setRealm(managerTestSetup.realmBuildingName) + createdAssetIds.add(childAsset.id) childAsset = assetStorageService.merge(childAsset) then: "another event should be sent" conditions.eventually { - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AssetEvent - assert (receivedEvents.get(0) as AssetEvent).id == childAsset.id - assert (receivedEvents.get(0) as AssetEvent).assetName == childAsset.name + def event = assertSingleEvent(receivedEvents, AssetEvent) + assert event.id == childAsset.id + assert event.assetName == childAsset.name } receivedEvents.clear() when: "the mqtt client unsubscribes from the multilevel topic" client.removeMessageConsumer(topic, eventConsumer) - then: "only two subscriptions should be left" + then: "only the realm children subscription should be left" conditions.eventually { - assert client.topicConsumerMap.get(topic) == null - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 2 + assertUnsubscribed(client, topic, keycloakTestSetup.serviceUser.id, 1) } when: "the descendant asset is modified" childAsset.addAttributes(new Attribute<>("temp2", TEXT)) childAsset = assetStorageService.merge(childAsset) - then: "A publish event message should not be sent" - new PollingConditions(initialDelay: 1, delay: 1, timeout: 10).eventually { - assert receivedEvents.size() == 0 + then: "a publish event message should not be sent" + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null + } + + def "MQTT last wills update authorised attributes and clients can publish after reconnect"() { + given: "an independent observer subscribed to the last will attribute" + preserveAttributes(managerTestSetup.apartment1HallwayId) + def observerId = UniqueIdentifierGenerator.generateId() + def observer = connectMqttClient(observerId) + def observerTopic = "${keycloakTestSetup.realmBuilding.name}/$observerId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() + observer.addMessageConsumer(observerTopic, eventConsumer) + conditions.eventually { + assertSubscribed(observer, observerTopic, keycloakTestSetup.serviceUser.id, 1) } + def mqttClientId = UniqueIdentifierGenerator.generateId() when: "another client connects with a last will configured to write to an attribute" - newClientId = "newClient" - def username2 = keycloakTestSetup.realmBuilding.name + ":" + keycloakTestSetup.serviceUser2.username // realm and OAuth client id - def password2 = keycloakTestSetup.serviceUser2.secret - newClient = new MQTT_IOClient(newClientId, mqttHost, mqttPort, false, true, new UsernamePassword(username2, password2), null, new MQTTLastWill("${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_VALUE_WRITE_TOPIC/motionSensor/$managerTestSetup.apartment1HallwayId".toString(), "1000", false)) - newClient.connect() + def client = createMqttClient(mqttClientId, restrictedMqttCredentials, new MQTTLastWill("${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_VALUE_WRITE_TOPIC/motionSensor/$managerTestSetup.apartment1HallwayId".toString(), "1000", false)) + client.connect() then: "the client should be connected" conditions.eventually { - assert newClient.getConnectionStatus() == ConnectionStatus.CONNECTED + assert client.getConnectionStatus() == ConnectionStatus.CONNECTED } - when: "the new client publishes to an attribute topic" + when: "the client publishes to an attribute topic" receivedEvents.clear() - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_VALUE_WRITE_TOPIC/motionSensor/${managerTestSetup.apartment1HallwayId}".toString() - payload = "170" - newClient.sendMessage(new MQTTMessage(topic, payload)) + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_VALUE_WRITE_TOPIC/motionSensor/${managerTestSetup.apartment1HallwayId}".toString() + def payload = "170" + client.sendMessage(new MQTTMessage(topic, payload)) - then: "The value of the attribute should be updated and the first client should have received the event" + then: "the attribute should be updated and the observer should receive the event" new PollingConditions(initialDelay: 1, timeout: 10, delay: 1).eventually { assert assetStorageService.find(managerTestSetup.apartment1HallwayId).getAttribute("motionSensor").get().value.orElse(0) == 170d - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 170d + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 170d) } receivedEvents.clear() - and: "the new client gets abruptly disconnected" + when: "the client's connection is closed abruptly" widenDisconnectedConnectionWindow(mqttBrokerService) def existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - // ((NioSocketChannel)((MqttClientConnectionConfig)((MqttClientConfig)((Mqtt3ClientConfigView)((Mqtt3AsyncClientView)device1Client.client).clientConfig).delegate).connectionConfig.get()).channel).config().setOption(ChannelOption.SO_LINGER, 0I) - ((SocketChannel)((MqttClientConnectionConfig)((MqttClientConfig)((Mqtt3ClientConfigView)((Mqtt3AsyncClientView)newClient.client).clientConfig).delegate).connectionConfig.get()).channel).close() + closeClientChannel(client) then: "the client should reconnect" conditions.eventually { - assert !mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).isEmpty() - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] !== existingConnection + assertReconnected(client, keycloakTestSetup.serviceUser2.id, existingConnection) } - and: "The last will message should have updated the value of the attribute and the first client should have received the event" - // The client reconnecting does not mean the broker has processed the closed session yet, and publishing the - // last will of that session is what triggers the update, so allow well beyond the reconnect for it + and: "the last will should update the attribute and the observer should receive the event" + // The broker processes the last will asynchronously, so verify its effect separately from the reconnect. new PollingConditions(initialDelay: 1, timeout: 30, delay: 1).eventually { assert assetStorageService.find(managerTestSetup.apartment1HallwayId).getAttribute("motionSensor").get().value.orElse(0) == 1000d - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 1000d + assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 1000d) } - when: "the new client tries to publish again with a specified timestamp" - def specifiedTimestamp = (receivedEvents.get(0) as AttributeEvent).timestamp + 5 + when: "the client publishes again with a specified timestamp" + def specifiedTimestamp = (receivedEvents.peek() as AttributeEvent).timestamp + 5 receivedEvents.clear() - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_WRITE_TOPIC/motionSensor/${managerTestSetup.apartment1HallwayId}".toString() + topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_WRITE_TOPIC/motionSensor/${managerTestSetup.apartment1HallwayId}".toString() payload = "{\"value\": 170, \"timestamp\": ${specifiedTimestamp}}".toString() - newClient.sendMessage(new MQTTMessage(topic, payload)) + client.sendMessage(new MQTTMessage(topic, payload)) - then: "The value of the attribute should be updated and the first client should have received the event" + then: "the attribute should be updated and the observer should receive the event with that timestamp" new PollingConditions(initialDelay: 1, timeout: 10, delay: 1).eventually { assert assetStorageService.find(managerTestSetup.apartment1HallwayId).getAttribute("motionSensor").get().value.orElse(0) == 170d - assert receivedEvents.size() == 1 - assert receivedEvents.get(0) instanceof AttributeEvent - assert (receivedEvents.get(0) as AttributeEvent).id == managerTestSetup.apartment1HallwayId - assert (receivedEvents.get(0) as AttributeEvent).name == "motionSensor" - assert (receivedEvents.get(0) as AttributeEvent).value.orElse(0) == 170d - assert (receivedEvents.get(0) as AttributeEvent).timestamp == specifiedTimestamp + def event = assertAttributeEvent(receivedEvents, managerTestSetup.apartment1HallwayId, "motionSensor", 170d) + assert event.timestamp == specifiedTimestamp } receivedEvents.clear() - when: "the new client disconnects" - newClient.disconnect() + when: "the client disconnects" + client.disconnect() then: "the status should show as disconnected" conditions.eventually { - assert newClient.getConnectionStatus() == ConnectionStatus.DISCONNECTED + assert client.getConnectionStatus() == ConnectionStatus.DISCONNECTED assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).isEmpty() } + } + + def "MQTT last wills cannot update unauthorised attributes"() { + given: "a client ID for the restricted user" + preserveAttributes(managerTestSetup.apartment2LivingroomId) + def mqttClientId = UniqueIdentifierGenerator.generateId() - when: "the new client reconnects using a last will topic for an unauthorised asset/attribute" + when: "a restricted client connects with a last will topic for an unauthorised asset/attribute" def asset = assetStorageService.find(managerTestSetup.apartment2LivingroomId) def currentCO2Level = asset.getAttribute("co2Level").flatMap{it.getValue()}.orElse(0d) - newClient = new MQTT_IOClient(newClientId, mqttHost, mqttPort, false, true, new UsernamePassword(username2, password2), null, new MQTTLastWill("${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_VALUE_WRITE_TOPIC/co2Level/$managerTestSetup.apartment2LivingroomId".toString(), "1000", false)) - newClient.setTopicSubscribeFailureConsumer {subFailures.add(it)} - newClient.connect() + def client = createMqttClient(mqttClientId, restrictedMqttCredentials, new MQTTLastWill("${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_VALUE_WRITE_TOPIC/co2Level/$managerTestSetup.apartment2LivingroomId".toString(), "1000", false)) + client.connect() then: "the client should be connected" conditions.eventually { - assert newClient.getConnectionStatus() == ConnectionStatus.CONNECTED + assert client.getConnectionStatus() == ConnectionStatus.CONNECTED assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 } - when: "the new client gets abruptly disconnected" - existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - // ((NioSocketChannel)((MqttClientConnectionConfig)((MqttClientConfig)((Mqtt3ClientConfigView)((Mqtt3AsyncClientView)device1Client.client).clientConfig).delegate).connectionConfig.get()).channel).config().setOption(ChannelOption.SO_LINGER, 0I) - ((SocketChannel)((MqttClientConnectionConfig)((MqttClientConfig)((Mqtt3ClientConfigView)((Mqtt3AsyncClientView)newClient.client).clientConfig).delegate).connectionConfig.get()).channel).close() + when: "the client's connection is closed abruptly" + widenDisconnectedConnectionWindow(mqttBrokerService) + def existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] + closeClientChannel(client) then: "the client should reconnect" conditions.eventually { - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] !== existingConnection + assertReconnected(client, keycloakTestSetup.serviceUser2.id, existingConnection) } - then: "The last will message should not have updated the value of the attribute" + then: "the last will message should not have updated the value of the attribute" new PollingConditions(initialDelay: 1, timeout: 10, delay: 1).eventually { def asst = assetStorageService.find(managerTestSetup.apartment2LivingroomId) assert asst.getAttribute("co2Level").get().value.orElse(-100) == currentCO2Level } + } + + def "MQTT restricted subscriptions are retried when user asset links change"() { + given: "a connected restricted client that retains failed subscriptions" + preserveUserAssetLinks() + def mqttClientId = UniqueIdentifierGenerator.generateId() + List subFailures = new CopyOnWriteArrayList<>() + def client = createMqttClient(mqttClientId, restrictedMqttCredentials) + client.setTopicSubscribeFailureConsumer { subFailures.add(it) } + client.connect() + conditions.eventually { + assert client.getConnectionStatus() == ConnectionStatus.CONNECTED + assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 + } when: "a restricted mqtt client subscribes to an unlinked asset" - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1BathroomId".toString() - newClient.addMessageConsumer(topic, {msg ->}) + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1BathroomId".toString() + def bathroomTopic = topic + client.addMessageConsumer(topic, {msg ->}) then: "the subscription should fail but the consumer should still exist (as setRemoveConsumersOnSubscriptionFailure=false)" conditions.eventually { - assert subFailures.contains(topic) - assert newClient.topicConsumerMap.get(topic) != null - assert newClient.topicConsumerMap.get(topic).consumers.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertSubscriptionFailed(client, topic, subFailures) + assertNoBrokerSubscriptions(keycloakTestSetup.serviceUser2.id) } when: "a restricted mqtt client subscribes to a linked asset" - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() - newClient.addMessageConsumer(topic, {msg ->}) + topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() + def hallwayTopic = topic + client.addMessageConsumer(topic, {msg ->}) then: "a subscription should exist" conditions.eventually { - assert newClient.topicConsumerMap.get(topic) != null - assert newClient.topicConsumerMap.get(topic).consumers.size() == 1 - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) - assert defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connection)).size() == 1 + assertSubscribed(client, topic, keycloakTestSetup.serviceUser2.id, 1) } when: "a user asset link is added for a connected restricted user" - existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] + def existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] assetStorageService.storeUserAssetLinks(List.of( new UserAssetLink(keycloakTestSetup.realmBuilding.getName(), keycloakTestSetup.serviceUser2.getId(), @@ -672,14 +688,14 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the existing connection should be terminated and the client should reconnect and the previously failed subscription should now succeed" conditions.eventually { - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] !== existingConnection - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertReconnected(client, keycloakTestSetup.serviceUser2.id, existingConnection) + assertSubscribed(client, bathroomTopic, keycloakTestSetup.serviceUser2.id) + assertSubscribed(client, hallwayTopic, keycloakTestSetup.serviceUser2.id) } when: "a user asset link is removed for a connected restricted user" existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] + subFailures.clear() assetStorageService.deleteUserAssetLinks(List.of( new UserAssetLink(keycloakTestSetup.realmBuilding.getName(), keycloakTestSetup.serviceUser2.getId(), @@ -687,54 +703,68 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the existing connection should be terminated and the client should reconnect and the previously failed subscription should be tried again but also fail" conditions.eventually { - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] !== existingConnection + assertReconnected(client, keycloakTestSetup.serviceUser2.id, existingConnection) + assertSubscribed(client, bathroomTopic, keycloakTestSetup.serviceUser2.id) + assertSubscriptionFailed(client, hallwayTopic, subFailures) } when: "the restricted mqtt client removes all subscriptions" - newClient.removeAllMessageConsumers() // Clear subscriptions as otherwise it won't hit the server again + client.removeAllMessageConsumers() // Clear subscriptions as otherwise it won't hit the server again - then: "no subscriptions should exist in the client" - assert newClient.topicConsumerMap.isEmpty() - - Consumer> restrictedEventConsumer = { msg -> - def event = ValueUtil.parse(msg.payload as String, SharedEvent.class) - restrictedReceivedEvents.add(event.get()) + then: "no subscriptions should remain in the client or broker" + conditions.eventually { + assertNoSubscriptions(client, keycloakTestSetup.serviceUser2.id) } when: "the restricted mqtt client subscribes to a now unlinked asset" - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() - newClient.addMessageConsumer(topic, restrictedEventConsumer) + topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$managerTestSetup.apartment1HallwayId".toString() + subFailures.clear() + client.addMessageConsumer(topic, eventConsumer) then: "no subscription should exist" conditions.eventually { - assert subFailures.contains(topic) - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - assert !defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertSubscriptionFailed(client, topic, subFailures) + assertNoBrokerSubscriptions(keycloakTestSetup.serviceUser2.id) } when: "the previously failed message consumer is removed" - newClient.removeMessageConsumer(topic, restrictedEventConsumer) + client.removeMessageConsumer(topic, eventConsumer) - and: "a restricted mqtt client subscribes to the attribute wildcard topic" - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() - newClient.addMessageConsumer(topic, restrictedEventConsumer) + then: "the failed subscription should be removed from the client" + assertUnsubscribed(client, topic, keycloakTestSetup.serviceUser2.id) + } + + def "MQTT restricted attribute subscriptions filter asset links and restricted read metadata"() { + given: "a restricted client linked to the bathroom but not the hallway" + preserveAttributes(managerTestSetup.apartment1HallwayId, managerTestSetup.apartment1BathroomId) + preserveUserAssetLinks() + assetStorageService.storeUserAssetLinks(List.of( + new UserAssetLink(keycloakTestSetup.realmBuilding.name, + keycloakTestSetup.serviceUser2.id, managerTestSetup.apartment1BathroomId))) + assetStorageService.deleteUserAssetLinks(List.of( + new UserAssetLink(keycloakTestSetup.realmBuilding.name, + keycloakTestSetup.serviceUser2.id, managerTestSetup.apartment1HallwayId))) + def mqttClientId = UniqueIdentifierGenerator.generateId() + def client = connectMqttClient(mqttClientId, restrictedMqttCredentials) + + when: "a restricted mqtt client subscribes to the attribute wildcard topic" + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ATTRIBUTE_TOPIC/$MQTTHandler.TOKEN_SINGLE_LEVEL_WILDCARD/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() + client.addMessageConsumer(topic, eventConsumer) then: "a subscription should exist" conditions.eventually { - assert newClient.topicConsumerMap.get(topic) != null - def connection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] - assert defaultMQTTHandler.sessionSubscriptionConsumers.containsKey(getConnectionIDString(connection)) + assertSubscribed(client, topic, keycloakTestSetup.serviceUser2.id) } when: "an unlinked asset attribute event occurs" - attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "motionSensor", 40) + def attributeEvent = new AttributeEvent(managerTestSetup.apartment1HallwayId, "motionSensor", 40) assetProcessingService.sendAttributeEvent(attributeEvent) then: "the restricted client should not receive the event" conditions.eventually { - assert restrictedReceivedEvents.size() == 0 + assert assetStorageService.find(attributeEvent.id).getAttribute(attributeEvent.name).get().value == attributeEvent.value } + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null when: "a linked asset attribute event occurs" attributeEvent = new AttributeEvent(managerTestSetup.apartment1BathroomId, RoomAsset.ROOM_NUMBER, 1) @@ -742,8 +772,9 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the restricted client should not receive the event since the attribute is not restricted" conditions.eventually { - assert restrictedReceivedEvents.size() == 0 + assert assetStorageService.find(attributeEvent.id).getAttribute(attributeEvent.name).get().value == attributeEvent.value } + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null when: "a linked asset attribute has its restricted meta configuration set to true" def bathroomAsset = assetStorageService.find(managerTestSetup.apartment1BathroomId) @@ -752,15 +783,15 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the restricted client should still be connected" conditions.eventually { - assert newClient.getConnectionStatus() == ConnectionStatus.CONNECTED + assert client.getConnectionStatus() == ConnectionStatus.CONNECTED assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 } then: "the restricted client should receive the triggered attribute event" conditions.eventually { - assert restrictedReceivedEvents.size() == 1 + assert receivedEvents.size() == 1 } - restrictedReceivedEvents.clear() + receivedEvents.clear() when: "another attribute event occurs on the restricted read attribute" attributeEvent = new AttributeEvent(managerTestSetup.apartment1BathroomId, RoomAsset.ROOM_NUMBER, 2) @@ -768,10 +799,10 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the restricted client should receive the triggered attribute event" conditions.eventually { - assert restrictedReceivedEvents.size() == 1 - assert restrictedReceivedEvents.get(0) == attributeEvent + assert receivedEvents.size() == 1 + assert receivedEvents.peek() == attributeEvent } - restrictedReceivedEvents.clear() + receivedEvents.clear() when: "the restricted read meta item is removed" bathroomAsset = assetStorageService.find(managerTestSetup.apartment1BathroomId) @@ -779,19 +810,31 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { assetStorageService.merge(bathroomAsset) then: "the restricted client should not receive the triggered attribute event" - conditions.eventually { - assert restrictedReceivedEvents.size() == 0 - } - newClient.removeAllMessageConsumers() - restrictedReceivedEvents.clear() + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null + client.removeAllMessageConsumers() + receivedEvents.clear() + } + + def "MQTT restricted asset subscriptions filter attributes and stop after unlinking"() { + given: "a restricted client linked to an asset with an attribute without restricted read access" + preserveAttributes(managerTestSetup.apartment1BathroomId) + preserveUserAssetLinks() + assetStorageService.storeUserAssetLinks(List.of( + new UserAssetLink(keycloakTestSetup.realmBuilding.name, + keycloakTestSetup.serviceUser2.id, managerTestSetup.apartment1BathroomId))) + def bathroomAsset = assetStorageService.find(managerTestSetup.apartment1BathroomId) + bathroomAsset.attributes.addOrReplace(new Attribute<>(RoomAsset.ROOM_NUMBER, 2)) + assetStorageService.merge(bathroomAsset) + def mqttClientId = UniqueIdentifierGenerator.generateId() + def client = connectMqttClient(mqttClientId, restrictedMqttCredentials) when: "a restricted mqtt client subscribes to the asset wildcard topic" - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ASSET_TOPIC/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() - newClient.addMessageConsumer(topic, restrictedEventConsumer) + def topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ASSET_TOPIC/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() + client.addMessageConsumer(topic, eventConsumer) then: "a subscription should exist" conditions.eventually { - assert newClient.topicConsumerMap.get(topic) != null + assertSubscribed(client, topic, keycloakTestSetup.serviceUser2.id) } when: "a linked asset event occurs" @@ -801,18 +844,15 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the restricted client should receive the triggered asset event" conditions.eventually { - assert restrictedReceivedEvents.size() == 1 + assert receivedEvents.size() == 1 } and: "the event should only contain attributes that are read restricted" conditions.eventually { - assert restrictedReceivedEvents.size() == 1 - def event = restrictedReceivedEvents.get(0) - assert event instanceof AssetEvent - def assetEvent = event as AssetEvent - assert !assetEvent.getAsset().hasAttribute(RoomAsset.ROOM_NUMBER) + def event = assertSingleEvent(receivedEvents, AssetEvent) + assert !event.asset.hasAttribute(RoomAsset.ROOM_NUMBER) } - restrictedReceivedEvents.clear() + receivedEvents.clear() when: "the restricted read meta item is added again" bathroomAsset = assetStorageService.find(managerTestSetup.apartment1BathroomId) @@ -821,16 +861,13 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the restricted client should receive the triggered asset event" conditions.eventually { - assert restrictedReceivedEvents.size() == 1 - def event = restrictedReceivedEvents.get(0) - assert event instanceof AssetEvent - def assetEvent = event as AssetEvent - assert assetEvent.getAsset().getAttribute(RoomAsset.ROOM_NUMBER).isPresent() - assert assetEvent.getAsset().getAttribute(RoomAsset.ROOM_NUMBER).get().getValue().orElse(null) == 2 + def event = assertSingleEvent(receivedEvents, AssetEvent) + assert event.asset.getAttribute(RoomAsset.ROOM_NUMBER).flatMap { it.value }.orElse(null) == 2 } - restrictedReceivedEvents.clear() + receivedEvents.clear() when: "the asset is unlinked" + def existingConnection = mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] assetStorageService.deleteUserAssetLinks(List.of( new UserAssetLink(keycloakTestSetup.realmBuilding.getName(), keycloakTestSetup.serviceUser2.getId(), @@ -839,23 +876,25 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { then: "the existing connection should be terminated and the client should reconnect" conditions.eventually { - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).size() == 1 - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id)[0] !== existingConnection + assertReconnected(client, keycloakTestSetup.serviceUser2.id, existingConnection) + assertSubscribed(client, topic, keycloakTestSetup.serviceUser2.id) } when: "the restricted mqtt client removes all subscriptions" - newClient.removeAllMessageConsumers() // Clear subscriptions as otherwise it won't hit the server again + client.removeAllMessageConsumers() // Clear subscriptions as otherwise it won't hit the server again - then: "no subscriptions should exist in the client" - assert newClient.topicConsumerMap.isEmpty() + then: "no subscriptions should remain in the client or broker" + conditions.eventually { + assertNoSubscriptions(client, keycloakTestSetup.serviceUser2.id) + } when: "the restricted mqtt client subscribes to the asset wildcard topic again" - topic = "${keycloakTestSetup.realmBuilding.name}/$newClientId/$DefaultMQTTHandler.ASSET_TOPIC/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() - newClient.addMessageConsumer(topic, restrictedEventConsumer) + topic = "${keycloakTestSetup.realmBuilding.name}/$mqttClientId/$DefaultMQTTHandler.ASSET_TOPIC/$MQTTHandler.TOKEN_MULTI_LEVEL_WILDCARD".toString() + client.addMessageConsumer(topic, eventConsumer) then: "a subscription should exist" conditions.eventually { - assert newClient.topicConsumerMap.get(topic) != null + assertSubscribed(client, topic, keycloakTestSetup.serviceUser2.id) } when: "an asset event occurs on the now unlinked asset" @@ -864,37 +903,141 @@ class MqttBrokerTest extends Specification implements ManagerContainerTrait { assetStorageService.merge(bathroomAsset) then: "the restricted client should not receive the triggered asset event" - conditions.eventually { - assert restrictedReceivedEvents.size() == 0 + assert receivedEvents.poll(1, TimeUnit.SECONDS) == null + receivedEvents.clear() + client.removeAllMessageConsumers() + } + + private void preserveAttributes(String... assetIds) { + assetIds.each { assetId -> + originalAttributes.putIfAbsent(assetId, ValueUtil.clone(assetStorageService.find(assetId).attributes)) } - restrictedReceivedEvents.clear() - newClient.removeAllMessageConsumers() + } - when: "both MQTT clients disconnect" - client.disconnect() - newClient.disconnect() + private void preserveUserAssetLinks() { + originalUserAssetLinks = assetStorageService.findUserAssetLinks(keycloakTestSetup.realmBuilding.name, + keycloakTestSetup.serviceUser2.id, null as String) + } - then: "no connections should be left" + private MQTT_IOClient createMqttClient(String clientId, UsernamePassword credentials = null, + MQTTLastWill lastWill = null) { + def client = new MQTT_IOClient(clientId, mqttHost, mqttPort, false, true, + credentials ?: mqttCredentials, null, lastWill) + clients.add(client) + return client + } + + private MQTT_IOClient connectMqttClient(String clientId, UsernamePassword credentials = null) { + def client = createMqttClient(clientId, credentials) + client.connect() conditions.eventually { - assert client.getConnectionStatus() == ConnectionStatus.DISCONNECTED - assert newClient.getConnectionStatus() == ConnectionStatus.DISCONNECTED - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser2.id).isEmpty() - assert mqttBrokerService.getUserConnections(keycloakTestSetup.serviceUser.id).isEmpty() + assert client.getConnectionStatus() == ConnectionStatus.CONNECTED } + return client + } - cleanup: "disconnect the clients" - if (client != null) { - client.disconnect() + private static Consumer> eventConsumerFor(Queue events) { + return { msg -> events.add(ValueUtil.parse(msg.payload as String, SharedEvent.class).get()) } + } + + private static T assertSingleEvent(Queue events, Class eventType) { + assert events.size() == 1 + def event = events.peek() + assert eventType.isInstance(event) + return eventType.cast(event) + } + + private static AttributeEvent assertAttributeEvent(Queue events, String assetId, + String name, Object value) { + def event = assertSingleEvent(events, AttributeEvent) + assert event.id == assetId + assert event.name == name + assert event.value.orElse(null) == value + return event + } + + private static void closeClientChannel(MQTT_IOClient mqttClient) { + def clientView = mqttClient.client as Mqtt3AsyncClientView + def configView = clientView.clientConfig as Mqtt3ClientConfigView + def clientConfig = configView.delegate as MqttClientConfig + def connectionConfig = clientConfig.connectionConfig.get() as MqttClientConnectionConfig + (connectionConfig.channel as SocketChannel).close() + } + + private Map brokerSubscriptions(String userId) { + def connections = mqttBrokerService.getUserConnections(userId) + assert connections.size() == 1 + return defaultMQTTHandler.sessionSubscriptionConsumers.get(getConnectionIDString(connections[0])) + } + + private void assertNoBrokerSubscriptions(String userId) { + assert brokerSubscriptions(userId) == null + } + + private void assertNoSubscriptions(MQTT_IOClient mqttClient, String userId) { + synchronized (mqttClient.topicConsumerMap) { + assert mqttClient.topicConsumerMap.isEmpty() } - if (newClient != null) { - newClient.disconnect() + assertNoBrokerSubscriptions(userId) + } + + private void assertUnsubscribed(MQTT_IOClient mqttClient, String subscriptionTopic, String userId, + int expectedRemainingSubscriptions = 0) { + synchronized (mqttClient.topicConsumerMap) { + assert mqttClient.topicConsumerMap.get(subscriptionTopic) == null + assert mqttClient.topicConsumerMap.size() == expectedRemainingSubscriptions } + if (expectedRemainingSubscriptions == 0) { + assertNoBrokerSubscriptions(userId) + } else { + def subscriptions = brokerSubscriptions(userId) + assert subscriptions != null + assert !subscriptions.containsKey(subscriptionTopic) + assert subscriptions.size() == expectedRemainingSubscriptions + } + } + + private static void assertSubscriptionFailed(MQTT_IOClient mqttClient, String subscriptionTopic, + Collection failedTopics) { + // Reconnects can report a failure more than once, so only check that this topic was rejected. + assert failedTopics.contains(subscriptionTopic) + synchronized (mqttClient.topicConsumerMap) { + def subscription = mqttClient.topicConsumerMap.get(subscriptionTopic) + if (mqttClient.isRemoveConsumersOnSubscriptionFailure()) { + assert subscription == null + } else { + assert subscription != null + assert subscription.isSubDone() + assert subscription.isSubFailed() + assert subscription.consumers.size() == 1 + } + } + } + + private void assertSubscribed(MQTT_IOClient mqttClient, String subscriptionTopic, String userId, + Integer expectedSubscriptionCount = null) { + assert mqttClient.getConnectionStatus() == ConnectionStatus.CONNECTED + synchronized (mqttClient.topicConsumerMap) { + def subscription = mqttClient.topicConsumerMap.get(subscriptionTopic) + assert subscription != null + assert subscription.subAckFuture.getNow(false) + assert subscription.consumers.size() == 1 + } + def subscriptions = brokerSubscriptions(userId) + assert subscriptions?.containsKey(subscriptionTopic) + if (expectedSubscriptionCount != null) { + assert subscriptions.size() == expectedSubscriptionCount + } + } + + private void assertReconnected(MQTT_IOClient mqttClient, String userId, RemotingConnection previousConnection) { + assert mqttClient.getConnectionStatus() == ConnectionStatus.CONNECTED + def connections = mqttBrokerService.getUserConnections(userId) + assert connections.size() == 1 + assert connections[0] !== previousConnection } - /** - * A last will publish is only accepted while the broker can still resolve the closed connection, which it keeps for - * 3s, so a publish consumer that is held up for longer drops the will and no amount of polling recovers it. - */ + /** Allow slow CI consumers to process last wills without hitting the independent cache expiry. */ private static void widenDisconnectedConnectionWindow(MQTTBrokerService mqttBrokerService) { mqttBrokerService.@disconnectedConnectionCache = CacheBuilder.newBuilder() .maximumSize(10000)