git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2024 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.binding.mqtt.homeassistant.internal;
  14
  15 import java.lang.ref.WeakReference;
  16 import java.util.HashSet;
  17 import java.util.Set;
  18 import java.util.concurrent.CompletableFuture;
  19 import java.util.concurrent.ScheduledExecutorService;
  20 import java.util.concurrent.ScheduledFuture;
  21 import java.util.concurrent.TimeUnit;
  22 import java.util.stream.Collectors;
  23
  24 import org.eclipse.jdt.annotation.NonNullByDefault;
  25 import org.eclipse.jdt.annotation.Nullable;
  26 import org.openhab.binding.mqtt.generic.AvailabilityTracker;
  27 import org.openhab.binding.mqtt.generic.ChannelStateUpdateListener;
  28 import org.openhab.binding.mqtt.generic.utils.FutureCollector;
  29 import org.openhab.binding.mqtt.homeassistant.internal.component.AbstractComponent;
  30 import org.openhab.binding.mqtt.homeassistant.internal.component.ComponentFactory;
  31 import org.openhab.binding.mqtt.homeassistant.internal.exception.ConfigurationException;
  32 import org.openhab.binding.mqtt.homeassistant.internal.exception.UnsupportedComponentException;
  33 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
  34 import org.openhab.core.io.transport.mqtt.MqttMessageSubscriber;
  35 import org.openhab.core.thing.ThingUID;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 import com.google.gson.Gson;
  40
  41 /**
  42  * Responsible for subscribing to the HomeAssistant MQTT components wildcard topic, either
  43  * in a time limited discovery mode or as a background discovery.
  44  *
  45  * @author David Graeff - Initial contribution
  46  */
  47 @NonNullByDefault
  48 public class DiscoverComponents implements MqttMessageSubscriber {
  49     private final Logger logger = LoggerFactory.getLogger(DiscoverComponents.class);
  50     private final ThingUID thingUID;
  51     private final ScheduledExecutorService scheduler;
  52     private final ChannelStateUpdateListener updateListener;
  53     private final AvailabilityTracker tracker;
  54     private final boolean newStyleChannels;
  55
  56     protected final CompletableFuture<@Nullable Void> discoverFinishedFuture = new CompletableFuture<>();
  57     private final Gson gson;
  58
  59     private @Nullable ScheduledFuture<?> stopDiscoveryFuture;
  60     private WeakReference<@Nullable MqttBrokerConnection> connectionRef = new WeakReference<>(null);
  61     protected @Nullable ComponentDiscovered discoveredListener;
  62     private int discoverTime;
  63     private Set<String> topics = new HashSet<>();
  64
  65     /**
  66      * Implement this to get notified of new components
  67      */
  68     public static interface ComponentDiscovered {
  69         void componentDiscovered(HaID homeAssistantTopicID, AbstractComponent<?> component);
  70     }
  71
  72     /**
  73      * Create a new discovery object.
  74      *
  75      * @param thingUID The Thing UID to perform the discovery for.
  76      * @param scheduler A scheduler for timeouts
  77      * @param channelStateUpdateListener Channel update listener. Usually the handler.
  78      */
  79     public DiscoverComponents(ThingUID thingUID, ScheduledExecutorService scheduler,
  80             ChannelStateUpdateListener channelStateUpdateListener, AvailabilityTracker tracker, Gson gson,
  81             boolean newStyleChannels) {
  82         this.thingUID = thingUID;
  83         this.scheduler = scheduler;
  84         this.updateListener = channelStateUpdateListener;
  85         this.gson = gson;
  86         this.tracker = tracker;
  87         this.newStyleChannels = newStyleChannels;
  88     }
  89
  90     @Override
  91     public void processMessage(String topic, byte[] payload) {
  92         if (!topic.endsWith("/config")) {
  93             return;
  94         }
  95
  96         HaID haID = new HaID(topic);
  97         String config = new String(payload);
  98         AbstractComponent<?> component = null;
  99
 100         if (config.length() > 0) {
 101             try {
 102                 component = ComponentFactory.createComponent(thingUID, haID, config, updateListener, tracker, scheduler,
 103                         gson, newStyleChannels);
 104                 component.setConfigSeen();
 105
 106                 logger.trace("Found HomeAssistant component {}", haID);
 107
 108                 if (discoveredListener != null) {
 109                     discoveredListener.componentDiscovered(haID, component);
 110                 }
 111             } catch (UnsupportedComponentException e) {
 112                 logger.warn("HomeAssistant discover error: thing {} component type is unsupported: {}", haID.objectID,
 113                         haID.component);
 114             } catch (ConfigurationException e) {
 115                 logger.warn("HomeAssistant discover error: invalid configuration of thing {} component {}: {}",
 116                         haID.objectID, haID.component, e.getMessage());
 117             } catch (Exception e) {
 118                 logger.warn("HomeAssistant discover error: {}", e.getMessage());
 119             }
 120         } else {
 121             logger.warn("Configuration of HomeAssistant thing {} is empty", haID.objectID);
 122         }
 123     }
 124
 125     /**
 126      * Start a components discovery.
 127      *
 128      * <p>
 129      * We need to consider the case that the remote client is using node IDs
 130      * and also the case that no node IDs are used.
 131      * </p>
 132      *
 133      * @param connection A MQTT broker connection
 134      * @param discoverTime The time in milliseconds for the discovery to run. Can be 0 to disable the
 135      *            timeout.
 136      *            You need to call {@link #stopDiscovery()} at some
 137      *            point in that case.
 138      * @param topicDescriptions Contains the object-id (=device id) and potentially a node-id as well.
 139      * @param componentsDiscoveredListener Listener for results
 140      * @return A future that completes normally after the given time in milliseconds or exceptionally on any error.
 141      *         Completes immediately if the timeout is disabled.
 142      */
 143     public CompletableFuture<@Nullable Void> startDiscovery(MqttBrokerConnection connection, int discoverTime,
 144             Set<HaID> topicDescriptions, ComponentDiscovered componentsDiscoveredListener) {
 145         this.topics = topicDescriptions.stream().map(id -> id.getTopic("config")).collect(Collectors.toSet());
 146         this.discoverTime = discoverTime;
 147         this.discoveredListener = componentsDiscoveredListener;
 148         this.connectionRef = new WeakReference<>(connection);
 149
 150         // Subscribe to the wildcard topic and start receive MQTT retained topics
 151         this.topics.stream().map(t -> connection.subscribe(t, this)).collect(FutureCollector.allOf())
 152                 .thenRun(this::subscribeSuccess).exceptionally(this::subscribeFail);
 153
 154         return discoverFinishedFuture;
 155     }
 156
 157     private void subscribeSuccess() {
 158         final MqttBrokerConnection connection = connectionRef.get();
 159         // Set up a scheduled future that will stop the discovery after the given time
 160         if (connection != null && discoverTime > 0) {
 161             this.stopDiscoveryFuture = scheduler.schedule(() -> {
 162                 this.stopDiscoveryFuture = null;
 163                 this.topics.stream().forEach(t -> connection.unsubscribe(t, this));
 164                 this.discoveredListener = null;
 165                 discoverFinishedFuture.complete(null);
 166             }, discoverTime, TimeUnit.MILLISECONDS);
 167         } else {
 168             // No timeout -> complete immediately
 169             discoverFinishedFuture.complete(null);
 170         }
 171     }
 172
 173     private @Nullable Void subscribeFail(Throwable e) {
 174         final ScheduledFuture<?> scheduledFuture = this.stopDiscoveryFuture;
 175         if (scheduledFuture != null) { // Cancel timeout
 176             scheduledFuture.cancel(false);
 177             this.stopDiscoveryFuture = null;
 178         }
 179         this.discoveredListener = null;
 180         final MqttBrokerConnection connection = connectionRef.get();
 181         if (connection != null) {
 182             this.topics.stream().forEach(t -> connection.unsubscribe(t, this));
 183             connectionRef.clear();
 184         }
 185         discoverFinishedFuture.completeExceptionally(e);
 186         return null;
 187     }
 188
 189     /**
 190      * Stops an ongoing discovery or do nothing if no discovery is running.
 191      */
 192     public void stopDiscovery() {
 193         subscribeFail(new Throwable("Stopped"));
 194     }
 195 }