[mqtt.homeassistant] Only subscribe to topics for linked channels (openhab#18239)

* [mqtt.homeassistant] Only subscribe to topics for linked channels

Signed-off-by: Cody Cutrer <[email protected]>

Author: ccutrer

bundles/org.openhab.binding.mqtt.homeassistant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/DiscoverComponents.java

@@ -52,6 +52,7 @@ public class DiscoverComponents implements MqttMessageSubscriber {
     private final ThingUID thingUID;
     private final ScheduledExecutorService scheduler;
     private final ChannelStateUpdateListener updateListener;
+    private final HomeAssistantChannelLinkageChecker linkageChecker;
     private final AvailabilityTracker tracker;
 
     protected final CompletableFuture<@Nullable Void> discoverFinishedFuture = new CompletableFuture<>();
@@ -82,11 +83,12 @@ public static interface ComponentDiscovered {
      * @param channelStateUpdateListener Channel update listener. Usually the handler.
      */
     public DiscoverComponents(ThingUID thingUID, ScheduledExecutorService scheduler,
-            ChannelStateUpdateListener channelStateUpdateListener, AvailabilityTracker tracker, Gson gson,
-            Jinjava jinjava, UnitProvider unitProvider) {
+            ChannelStateUpdateListener channelStateUpdateListener, HomeAssistantChannelLinkageChecker linkageChecker,
+            AvailabilityTracker tracker, Gson gson, Jinjava jinjava, UnitProvider unitProvider) {
         this.thingUID = thingUID;
         this.scheduler = scheduler;
         this.updateListener = channelStateUpdateListener;
+        this.linkageChecker = linkageChecker;
         this.gson = gson;
         this.jinjava = jinjava;
         this.unitProvider = unitProvider;
@@ -105,8 +107,8 @@ public void processMessage(String topic, byte[] payload) {
 
         if (config.length() > 0) {
             try {
-                component = ComponentFactory.createComponent(thingUID, haID, config, updateListener, tracker, scheduler,
-                        gson, jinjava, unitProvider);
+                component = ComponentFactory.createComponent(thingUID, haID, config, updateListener, linkageChecker,
+                        tracker, scheduler, gson, jinjava, unitProvider);
                 component.setConfigSeen();
 
                 logger.trace("Found HomeAssistant component {}", haID);

bundles/org.openhab.binding.mqtt.homeassistant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/HomeAssistantChannelLinkageChecker.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2010-2025 Contributors to the openHAB project
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.openhab.binding.mqtt.homeassistant.internal;
+
+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.openhab.core.thing.ChannelUID;
+
+/**
+ * Allows a Home Assistant component to check if a channel is linked (and thus worth subscribing to)
+ *
+ * @author Cody Cutrer - Initial contribution
+ */
+@NonNullByDefault
+public interface HomeAssistantChannelLinkageChecker {
+    /**
+     * Returns whether at least one item is linked for the given channel ID.
+     *
+     * @param channelId channel ID (must not be null)
+     * @return true if at least one item is linked, false otherwise
+     */
+    boolean isChannelLinked(ChannelUID channelUID);
+}

bundles/org.openhab.binding.mqtt.homeassistant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/component/AbstractComponent.java

@@ -55,6 +55,7 @@
 import org.openhab.core.thing.type.ChannelGroupType;
 import org.openhab.core.thing.type.ChannelGroupTypeBuilder;
 import org.openhab.core.thing.type.ChannelGroupTypeUID;
+import org.openhab.core.thing.type.ChannelKind;
 import org.openhab.core.types.CommandDescription;
 import org.openhab.core.types.StateDescription;
 
@@ -232,8 +233,14 @@ public void setConfigSeen() {
      */
     public CompletableFuture<@Nullable Void> start(MqttBrokerConnection connection, ScheduledExecutorService scheduler,
             int timeout) {
-        return Stream.concat(channels.values().stream(), hiddenChannels.stream())
-                .map(v -> v.start(connection, scheduler, timeout)) //
+        // Hidden channels (used by a component to simulate other channels or non-channel behavior),
+        // triggers channels (which can be used by rules without ever being linked),
+        // and linked channels are started.
+        // Therefore, unlinked channels are not started.
+        return Stream.concat(channels.values().stream().filter(c -> {
+            return c.getChannel().getKind().equals(ChannelKind.TRIGGER)
+                    || componentConfiguration.getLinkageChecker().isChannelLinked(c.getChannel().getUID());
+        }), hiddenChannels.stream()).map(v -> v.start(connection, scheduler, timeout)) //
                 .reduce(CompletableFuture.completedFuture(null), (f, v) -> f.thenCompose(b -> v));
     }
 

bundles/org.openhab.binding.mqtt.homeassistant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/component/ComponentFactory.java

@@ -18,6 +18,7 @@
 import org.openhab.binding.mqtt.generic.AvailabilityTracker;
 import org.openhab.binding.mqtt.generic.ChannelStateUpdateListener;
 import org.openhab.binding.mqtt.homeassistant.internal.HaID;
+import org.openhab.binding.mqtt.homeassistant.internal.HomeAssistantChannelLinkageChecker;
 import org.openhab.binding.mqtt.homeassistant.internal.config.dto.AbstractChannelConfiguration;
 import org.openhab.binding.mqtt.homeassistant.internal.exception.ConfigurationException;
 import org.openhab.binding.mqtt.homeassistant.internal.exception.UnsupportedComponentException;
@@ -47,10 +48,12 @@ public class ComponentFactory {
      * @return A HA MQTT Component
      */
     public static AbstractComponent<?> createComponent(ThingUID thingUID, HaID haID, String channelConfigurationJSON,
-            ChannelStateUpdateListener updateListener, AvailabilityTracker tracker, ScheduledExecutorService scheduler,
-            Gson gson, Jinjava jinjava, UnitProvider unitProvider) throws ConfigurationException {
+            ChannelStateUpdateListener updateListener, HomeAssistantChannelLinkageChecker linkageChecker,
+            AvailabilityTracker tracker, ScheduledExecutorService scheduler, Gson gson, Jinjava jinjava,
+            UnitProvider unitProvider) throws ConfigurationException {
         ComponentConfiguration componentConfiguration = new ComponentConfiguration(thingUID, haID,
-                channelConfigurationJSON, gson, jinjava, updateListener, tracker, scheduler, unitProvider);
+                channelConfigurationJSON, gson, jinjava, updateListener, linkageChecker, tracker, scheduler,
+                unitProvider);
         switch (haID.component) {
             case "alarm_control_panel":
                 return new AlarmControlPanel(componentConfiguration);
@@ -110,6 +113,7 @@ protected static class ComponentConfiguration {
         private final HaID haID;
         private final String configJSON;
         private final ChannelStateUpdateListener updateListener;
+        private final HomeAssistantChannelLinkageChecker linkageChecker;
         private final AvailabilityTracker tracker;
         private final Gson gson;
         private final Jinjava jinjava;
@@ -125,14 +129,15 @@ protected static class ComponentConfiguration {
          * @param gson A Gson instance
          */
         protected ComponentConfiguration(ThingUID thingUID, HaID haID, String configJSON, Gson gson, Jinjava jinjava,
-                ChannelStateUpdateListener updateListener, AvailabilityTracker tracker,
-                ScheduledExecutorService scheduler, UnitProvider unitProvider) {
+                ChannelStateUpdateListener updateListener, HomeAssistantChannelLinkageChecker linkageChecker,
+                AvailabilityTracker tracker, ScheduledExecutorService scheduler, UnitProvider unitProvider) {
             this.thingUID = thingUID;
             this.haID = haID;
             this.configJSON = configJSON;
             this.gson = gson;
             this.jinjava = jinjava;
             this.updateListener = updateListener;
+            this.linkageChecker = linkageChecker;
             this.tracker = tracker;
             this.scheduler = scheduler;
             this.unitProvider = unitProvider;
@@ -154,6 +159,10 @@ public ChannelStateUpdateListener getUpdateListener() {
             return updateListener;
         }
 
+        public HomeAssistantChannelLinkageChecker getLinkageChecker() {
+            return linkageChecker;
+        }
+
         public Gson getGson() {
             return gson;
         }

bundles/org.openhab.binding.mqtt.homeassistant/src/main/java/org/openhab/binding/mqtt/homeassistant/internal/handler/HomeAssistantThingHandler.java

@@ -39,6 +39,7 @@
 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents.ComponentDiscovered;
 import org.openhab.binding.mqtt.homeassistant.internal.HaID;
 import org.openhab.binding.mqtt.homeassistant.internal.HandlerConfiguration;
+import org.openhab.binding.mqtt.homeassistant.internal.HomeAssistantChannelLinkageChecker;
 import org.openhab.binding.mqtt.homeassistant.internal.component.AbstractComponent;
 import org.openhab.binding.mqtt.homeassistant.internal.component.ComponentFactory;
 import org.openhab.binding.mqtt.homeassistant.internal.component.DeviceTrigger;
@@ -83,7 +84,7 @@
  */
 @NonNullByDefault
 public class HomeAssistantThingHandler extends AbstractMQTTThingHandler
-        implements ComponentDiscovered, Consumer<List<Object>> {
+        implements ComponentDiscovered, Consumer<List<Object>>, HomeAssistantChannelLinkageChecker {
     public static final String AVAILABILITY_CHANNEL = "availability";
     private static final Comparator<AbstractComponent<?>> COMPONENT_COMPARATOR = Comparator
             .comparing((AbstractComponent<?> component) -> component.hasGroup())
@@ -134,7 +135,7 @@ public HomeAssistantThingHandler(Thing thing, MqttChannelTypeProvider channelTyp
         this.unitProvider = unitProvider;
         this.attributeReceiveTimeout = attributeReceiveTimeout;
         this.delayedProcessing = new DelayedBatchProcessing<>(attributeReceiveTimeout, this, scheduler);
-        this.discoverComponents = new DiscoverComponents(thing.getUID(), scheduler, this, this, gson, jinjava,
+        this.discoverComponents = new DiscoverComponents(thing.getUID(), scheduler, this, this, this, gson, jinjava,
                 unitProvider);
     }
 
@@ -183,7 +184,7 @@ public void initialize() {
                 String channelConfigurationJSON = (String) channelConfig.get("config");
                 try {
                     AbstractComponent<?> component = ComponentFactory.createComponent(thingUID, haID,
-                            channelConfigurationJSON, this, this, scheduler, gson, jinjava, unitProvider);
+                            channelConfigurationJSON, this, this, this, scheduler, gson, jinjava, unitProvider);
                     if (typeID.equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
                         typeID = calculateThingTypeUID(component);
                     }
@@ -558,4 +559,36 @@ private List<Configuration> flattenChannelConfiguration(Configuration multiCompo
     Map<@Nullable String, AbstractComponent<?>> getComponents() {
         return haComponents;
     }
+
+    // For components to check if a channel is linked before starting them
+    @Override
+    public boolean isChannelLinked(ChannelUID channelUID) {
+        return isLinked(channelUID);
+    }
+
+    // A channel is newly linked; make sure it is started
+    @Override
+    public void channelLinked(ChannelUID channelUID) {
+        MqttBrokerConnection connection = this.connection;
+        // We haven't started at all yet.
+        if (connection == null) {
+            return;
+        }
+        synchronized (haComponents) {
+            haComponents.forEach((id, component) -> {
+                if (component.getChannels().stream().anyMatch(channel -> channel.getUID().equals(channelUID))) {
+                    component.start(connection, scheduler, attributeReceiveTimeout).exceptionally(e -> {
+                        updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, e.getMessage());
+                        return null;
+                    });
+                }
+            });
+        }
+        super.channelLinked(channelUID);
+    }
+
+    // Don't bother unsubscribing on unlink; it's a relatively rare operation during normal usage,
+    // and a decent amount of effort to make sure there aren't other links before stopping them, and
+    // making sure not to stop other channels that are still linked.
+    // A disable/re-enable of the thing will clear the subscriptions.
 }

bundles/org.openhab.binding.mqtt.homeassistant/src/test/java/org/openhab/binding/mqtt/homeassistant/internal/component/AbstractComponentTests.java

@@ -19,6 +19,7 @@
 import static org.mockito.Mockito.*;
 
 import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -43,10 +44,12 @@
 import org.openhab.binding.mqtt.homeassistant.internal.handler.HomeAssistantThingHandler;
 import org.openhab.core.i18n.UnitProvider;
 import org.openhab.core.library.types.HSBType;
+import org.openhab.core.thing.ChannelUID;
 import org.openhab.core.thing.Thing;
 import org.openhab.core.thing.ThingStatusInfo;
 import org.openhab.core.thing.binding.ThingHandlerCallback;
 import org.openhab.core.thing.type.AutoUpdatePolicy;
+import org.openhab.core.thing.type.ChannelKind;
 import org.openhab.core.thing.type.ChannelTypeRegistry;
 import org.openhab.core.types.Command;
 import org.openhab.core.types.State;
@@ -137,6 +140,32 @@ public void disposeThingHandler() {
         return Objects.requireNonNull(thingHandler.getDiscoveredComponent());
     }
 
+    /**
+     * Simulate linking an item to a channel, so that the handler knows it should subscribe to the relevant topics
+     *
+     * @param component component
+     * @param channelId channel
+     */
+    protected void linkChannel(AbstractComponent<@NonNull ? extends AbstractChannelConfiguration> component,
+            String channelId) {
+        var stateChannel = Objects.requireNonNull(component.getChannel(channelId));
+        thingHandler.linkChannel(stateChannel.getChannel().getUID());
+    }
+
+    /**
+     * Simulate linking an item to a all channels of a component, so that the handler knows it should subscribe to the
+     * relevant topics
+     *
+     * @param component component
+     */
+    protected void linkAllChannels(AbstractComponent<@NonNull ? extends AbstractChannelConfiguration> component) {
+        component.getChannels().forEach(c -> {
+            if (c.getKind().equals(ChannelKind.STATE)) {
+                thingHandler.linkChannel(c.getUID());
+            }
+        });
+    }
+
     /**
      * Assert channel topics, label and value class
      *
@@ -328,6 +357,7 @@ protected void sendCommand(AbstractComponent<@NonNull ? extends AbstractChannelC
     }
 
     protected static class LatchThingHandler extends HomeAssistantThingHandler {
+        private final Set<ChannelUID> linkedChannels = new HashSet<>();
         private @Nullable CountDownLatch latch;
         private @Nullable AbstractComponent<@NonNull ? extends AbstractChannelConfiguration> discoveredComponent;
 
@@ -356,5 +386,15 @@ public CountDownLatch createWaitForComponentDiscoveredLatch(int count) {
         public @Nullable AbstractComponent<@NonNull ? extends AbstractChannelConfiguration> getDiscoveredComponent() {
             return discoveredComponent;
         }
+
+        public void linkChannel(ChannelUID channelUID) {
+            linkedChannels.add(channelUID);
+            channelLinked(channelUID);
+        }
+
+        @Override
+        protected boolean isLinked(ChannelUID channelUID) {
+            return linkedChannels.contains(channelUID);
+        }
     }
 }

bundles/org.openhab.binding.mqtt.homeassistant/src/test/java/org/openhab/binding/mqtt/homeassistant/internal/component/AlarmControlPanelTests.java

@@ -71,6 +71,8 @@ public void testAlarmControlPanel() {
         assertChannel(component, AlarmControlPanel.STATE_CHANNEL_ID, "zigbee2mqtt/alarm/state",
                 "zigbee2mqtt/alarm/set/state", "alarm", TextValue.class);
 
+        linkAllChannels(component);
+
         publishMessage("zigbee2mqtt/alarm/state", "armed_home");
         assertState(component, AlarmControlPanel.STATE_CHANNEL_ID, new StringType("armed_home"));
         publishMessage("zigbee2mqtt/alarm/state", "armed_away");

bundles/org.openhab.binding.mqtt.homeassistant/src/test/java/org/openhab/binding/mqtt/homeassistant/internal/component/BinarySensorTests.java

@@ -70,6 +70,8 @@ public void test() throws InterruptedException {
         assertChannel(component, BinarySensor.SENSOR_CHANNEL_ID, "zigbee2mqtt/sensor/state", "", "onoffsensor",
                 OnOffValue.class);
 
+        linkAllChannels(component);
+
         publishMessage("zigbee2mqtt/sensor/state", "{ \"state\": \"ON_\" }");
         assertState(component, BinarySensor.SENSOR_CHANNEL_ID, OnOffType.ON);
         publishMessage("zigbee2mqtt/sensor/state", "{ \"state\": \"ON_\" }");
@@ -112,6 +114,8 @@ public void offDelayTest() {
                 """);
         // @formatter:on
 
+        linkAllChannels(component);
+
         publishMessage("zigbee2mqtt/sensor/state", "{ \"state\": \"ON_\" }");
         assertState(component, BinarySensor.SENSOR_CHANNEL_ID, OnOffType.ON);
 
@@ -150,6 +154,8 @@ public void expireAfterTest() {
                 """);
         // @formatter:on
 
+        linkAllChannels(component);
+
         publishMessage("zigbee2mqtt/sensor/state", "{ \"state\": \"OFF_\" }");
         assertState(component, BinarySensor.SENSOR_CHANNEL_ID, OnOffType.OFF);
 

bundles/org.openhab.binding.mqtt.homeassistant/src/test/java/org/openhab/binding/mqtt/homeassistant/internal/component/ButtonTests.java

@@ -62,6 +62,8 @@ public void testButton() {
         assertThat(Objects.requireNonNull(component.getChannel(Button.BUTTON_CHANNEL_ID)).getChannel()
                 .getAutoUpdatePolicy(), is(AutoUpdatePolicy.VETO));
 
+        linkAllChannels(component);
+
         assertThrows(IllegalArgumentException.class,
                 () -> component.getChannel(Button.BUTTON_CHANNEL_ID).getState().publishValue(new StringType("ON")));
         assertNothingPublished("esphome/single-car-gdo/button/restart/command");

bundles/org.openhab.binding.mqtt.homeassistant/src/test/java/org/openhab/binding/mqtt/homeassistant/internal/component/CameraTests.java

@@ -62,6 +62,8 @@ public void test() throws InterruptedException {
 
         assertChannel(component, Camera.CAMERA_CHANNEL_ID, "zigbee2mqtt/cam1/state", "", "cam1", ImageValue.class);
 
+        linkAllChannels(component);
+
         var imageBytes = getResourceAsByteArray("component/image.png");
         publishMessage("zigbee2mqtt/cam1/state", imageBytes);
         assertState(component, Camera.CAMERA_CHANNEL_ID, new RawType(imageBytes, "image/png"));