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

   1 /**
   2  * Copyright (c) 2010-2023 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.handler;
  14
  15 import java.util.Collection;
  16 import java.util.Collections;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.concurrent.CompletableFuture;
  20 import java.util.concurrent.TimeoutException;
  21
  22 import org.eclipse.jdt.annotation.NonNullByDefault;
  23 import org.eclipse.jdt.annotation.Nullable;
  24 import org.openhab.binding.mqtt.discovery.MQTTTopicDiscoveryParticipant;
  25 import org.openhab.binding.mqtt.discovery.TopicSubscribe;
  26 import org.openhab.binding.mqtt.internal.action.MQTTActions;
  27 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
  28 import org.openhab.core.io.transport.mqtt.MqttConnectionObserver;
  29 import org.openhab.core.io.transport.mqtt.MqttConnectionState;
  30 import org.openhab.core.thing.Bridge;
  31 import org.openhab.core.thing.Channel;
  32 import org.openhab.core.thing.ChannelUID;
  33 import org.openhab.core.thing.ThingStatus;
  34 import org.openhab.core.thing.ThingStatusDetail;
  35 import org.openhab.core.thing.binding.BaseBridgeHandler;
  36 import org.openhab.core.thing.binding.ThingHandlerService;
  37 import org.openhab.core.types.Command;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41 /**
  42  * This base implementation handles connection changes of the {@link MqttBrokerConnection}
  43  * and puts the Thing on or offline. It also provides a basic dispose() implementation.
  44  *
  45  * @author David Graeff - Initial contribution
  46  */
  47 @NonNullByDefault
  48 public abstract class AbstractBrokerHandler extends BaseBridgeHandler implements MqttConnectionObserver {
  49     public static final int TIMEOUT_DEFAULT = 1200; /* timeout in milliseconds */
  50     private final Logger logger = LoggerFactory.getLogger(AbstractBrokerHandler.class);
  51
  52     final Map<ChannelUID, PublishTriggerChannel> channelStateByChannelUID = new HashMap<>();
  53     private final Map<String, @Nullable Map<MQTTTopicDiscoveryParticipant, @Nullable TopicSubscribe>> discoveryTopics = new HashMap<>();
  54
  55     protected @Nullable MqttBrokerConnection connection;
  56     protected CompletableFuture<MqttBrokerConnection> connectionFuture = new CompletableFuture<>();
  57
  58     public AbstractBrokerHandler(Bridge thing) {
  59         super(thing);
  60     }
  61
  62     @Override
  63     public Collection<Class<? extends ThingHandlerService>> getServices() {
  64         return Collections.singleton(MQTTActions.class);
  65     }
  66
  67     /**
  68      * Returns the underlying {@link MqttBrokerConnection} either immediately or after {@link #initialize()} has
  69      * performed.
  70      */
  71     public CompletableFuture<MqttBrokerConnection> getConnectionAsync() {
  72         return connectionFuture;
  73     }
  74
  75     /**
  76      * Returns the underlying {@link MqttBrokerConnection}.
  77      */
  78     public @Nullable MqttBrokerConnection getConnection() {
  79         return connection;
  80     }
  81
  82     /**
  83      * Does nothing in the base implementation.
  84      */
  85     @Override
  86     public void handleCommand(ChannelUID channelUID, Command command) {
  87         // No commands to handle
  88     }
  89
  90     /**
  91      * Registers a connection status listener and attempts a connection if there is none so far.
  92      */
  93     @Override
  94     public void initialize() {
  95         final MqttBrokerConnection connection = this.connection;
  96         if (connection == null) {
  97             logger.warn("Trying to initialize {} but connection is null. This is most likely a bug.", thing.getUID());
  98             return;
  99         }
 100         for (Channel channel : thing.getChannels()) {
 101             final PublishTriggerChannelConfig channelConfig = channel.getConfiguration()
 102                     .as(PublishTriggerChannelConfig.class);
 103             PublishTriggerChannel c = new PublishTriggerChannel(channelConfig, channel.getUID(), connection, this);
 104             channelStateByChannelUID.put(channel.getUID(), c);
 105         }
 106
 107         connection.addConnectionObserver(this);
 108
 109         connection.start().exceptionally(e -> {
 110             connectionStateChanged(MqttConnectionState.DISCONNECTED, e);
 111             return false;
 112         }).thenAccept(v -> {
 113             if (!v) {
 114                 connectionStateChanged(MqttConnectionState.DISCONNECTED, new TimeoutException("Timeout"));
 115             }
 116         });
 117         connectionFuture.complete(connection);
 118
 119         discoveryTopics.forEach((topic, listenerMap) -> {
 120             listenerMap.replaceAll((listener, oldTopicSubscribe) -> {
 121                 if (oldTopicSubscribe.isStarted()) {
 122                     oldTopicSubscribe.stop();
 123                 }
 124
 125                 TopicSubscribe topicSubscribe = new TopicSubscribe(connection, topic, listener, thing.getUID());
 126                 if (discoveryEnabled()) {
 127                     topicSubscribe.start().handle((result, ex) -> {
 128                         if (ex != null) {
 129                             logger.warn("Failed to subscribe {} to discovery topic {} on broker {}", listener, topic,
 130                                     thing.getUID());
 131                         } else {
 132                             logger.trace("Subscribed {} to discovery topic {} on broker {}", listener, topic,
 133                                     thing.getUID());
 134                         }
 135                         return null;
 136                     });
 137                 }
 138                 return topicSubscribe;
 139             });
 140         });
 141     }
 142
 143     @Override
 144     public void connectionStateChanged(MqttConnectionState state, @Nullable Throwable error) {
 145         if (state == MqttConnectionState.CONNECTED) {
 146             updateStatus(ThingStatus.ONLINE);
 147             channelStateByChannelUID.values().forEach(PublishTriggerChannel::start);
 148         } else {
 149             channelStateByChannelUID.values().forEach(PublishTriggerChannel::stop);
 150             if (error == null) {
 151                 updateStatus(ThingStatus.OFFLINE);
 152             } else {
 153                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR, error.getMessage());
 154             }
 155         }
 156     }
 157
 158     @Override
 159     protected void triggerChannel(ChannelUID channelUID, String event) {
 160         super.triggerChannel(channelUID, event);
 161     }
 162
 163     /**
 164      * Removes listeners to the {@link MqttBrokerConnection}.
 165      */
 166     @Override
 167     public void dispose() {
 168         channelStateByChannelUID.values().forEach(PublishTriggerChannel::stop);
 169         channelStateByChannelUID.clear();
 170
 171         // keep topics, but stop subscriptions
 172         discoveryTopics.forEach((topic, listenerMap) -> {
 173             listenerMap.forEach((listener, topicSubscribe) -> {
 174                 topicSubscribe.stop();
 175             });
 176         });
 177
 178         if (connection != null) {
 179             connection.removeConnectionObserver(this);
 180         } else {
 181             logger.warn("Trying to dispose handler {} but connection is already null. Most likely this is a bug.",
 182                     thing.getUID());
 183         }
 184         this.connection = null;
 185         connectionFuture = new CompletableFuture<>();
 186         super.dispose();
 187     }
 188
 189     /**
 190      * register a discovery listener to a specified topic on this broker (used by the handler factory)
 191      *
 192      * @param listener the discovery participant that wishes to be notified about this topic
 193      * @param topic the topic (wildcards supported)
 194      */
 195     public final void registerDiscoveryListener(MQTTTopicDiscoveryParticipant listener, String topic) {
 196         Map<MQTTTopicDiscoveryParticipant, @Nullable TopicSubscribe> topicListeners = discoveryTopics
 197                 .computeIfAbsent(topic, t -> new HashMap<>());
 198         topicListeners.compute(listener, (k, v) -> {
 199             if (v != null) {
 200                 logger.warn("Duplicate subscription for {} to discovery topic {} on broker {}. Check discovery logic!",
 201                         listener, topic, thing.getUID());
 202                 v.stop();
 203             }
 204
 205             TopicSubscribe topicSubscribe = new TopicSubscribe(connection, topic, listener, thing.getUID());
 206             if (discoveryEnabled()) {
 207                 topicSubscribe.start().handle((result, ex) -> {
 208                     if (ex != null) {
 209                         logger.warn("Failed to subscribe {} to discovery topic {} on broker {}", listener, topic,
 210                                 thing.getUID());
 211                     } else {
 212                         logger.trace("Subscribed {} to discovery topic {} on broker {}", listener, topic,
 213                                 thing.getUID());
 214                     }
 215                     return null;
 216                 });
 217             }
 218             return topicSubscribe;
 219         });
 220     }
 221
 222     /**
 223      * unregisters a discovery listener from a specified topic on this broker (used by the handler factory)
 224      *
 225      * @param listener the discovery participant that wishes no notifications about this topic
 226      * @param topic the topic (as specified during registration)
 227      */
 228     public final void unregisterDiscoveryListener(MQTTTopicDiscoveryParticipant listener, String topic) {
 229         discoveryTopics.compute(topic, (k, v) -> {
 230             if (v == null) {
 231                 logger.warn(
 232                         "Tried to unsubscribe {} from  discovery topic {} on broker {} but topic not registered at all. Check discovery logic!",
 233                         listener, topic, thing.getUID());
 234                 return null;
 235             }
 236             v.compute(listener, (l, w) -> {
 237                 if (w == null) {
 238                     logger.warn(
 239                             "Tried to unsubscribe {} from  discovery topic {} on broker {} but topic not registered for listener. Check discovery logic!",
 240                             listener, topic, thing.getUID());
 241                 } else {
 242                     w.stop();
 243                     logger.trace("Unsubscribed {} from discovery topic {} on broker {}", listener, topic,
 244                             thing.getUID());
 245                 }
 246                 return null;
 247             });
 248             return v.isEmpty() ? null : v;
 249         });
 250     }
 251
 252     /**
 253      * check whether discovery is disabled on this broker
 254      *
 255      * @return true if discovery disabled
 256      */
 257     public abstract boolean discoveryEnabled();
 258 }