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