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.homie.internal.homie300;
  14
  15 import java.util.ArrayList;
  16 import java.util.List;
  17 import java.util.Objects;
  18 import java.util.concurrent.CompletableFuture;
  19 import java.util.concurrent.ScheduledExecutorService;
  20 import java.util.stream.Collectors;
  21 import java.util.stream.Stream;
  22
  23 import org.eclipse.jdt.annotation.NonNullByDefault;
  24 import org.eclipse.jdt.annotation.Nullable;
  25 import org.openhab.binding.mqtt.generic.mapping.AbstractMqttAttributeClass;
  26 import org.openhab.binding.mqtt.generic.tools.ChildMap;
  27 import org.openhab.binding.mqtt.homie.generic.internal.MqttBindingConstants;
  28 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
  29 import org.openhab.core.thing.ChannelGroupUID;
  30 import org.openhab.core.thing.ThingUID;
  31 import org.openhab.core.thing.type.ChannelDefinition;
  32 import org.openhab.core.thing.type.ChannelDefinitionBuilder;
  33 import org.openhab.core.thing.type.ChannelGroupType;
  34 import org.openhab.core.thing.type.ChannelGroupTypeBuilder;
  35 import org.openhab.core.thing.type.ChannelGroupTypeUID;
  36 import org.openhab.core.util.UIDUtils;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39
  40 /**
  41  * Homie 3.x Node.
  42  *
  43  * A Homie Node contains Homie Properties ({@link Property}) but can also have attributes ({@link NodeAttributes}).
  44  * It corresponds to a ChannelGroup.
  45  *
  46  * @author David Graeff - Initial contribution
  47  */
  48 @NonNullByDefault
  49 public class Node implements AbstractMqttAttributeClass.AttributeChanged {
  50     private final Logger logger = LoggerFactory.getLogger(Node.class);
  51     // Homie
  52     public final String nodeID;
  53     public final NodeAttributes attributes;
  54     public ChildMap<Property> properties;
  55     // Runtime
  56     public final DeviceCallback callback;
  57     protected final ChannelGroupUID channelGroupUID;
  58     public final ChannelGroupTypeUID channelGroupTypeUID;
  59     private final String topic;
  60     private boolean initialized = false;
  61
  62     /**
  63      * Creates a Homie Node.
  64      *
  65      * @param topic The base topic for this node (e.g. "homie/device")
  66      * @param nodeID The node ID
  67      * @param thingUID The Thing UID, used to determine the ChannelGroupUID.
  68      * @param callback The callback for the handler.
  69      */
  70     public Node(String topic, String nodeID, ThingUID thingUID, DeviceCallback callback, NodeAttributes attributes) {
  71         this.attributes = attributes;
  72         this.topic = topic + "/" + nodeID;
  73         this.nodeID = nodeID;
  74         this.callback = callback;
  75         channelGroupTypeUID = new ChannelGroupTypeUID(MqttBindingConstants.BINDING_ID, UIDUtils.encode(this.topic));
  76         channelGroupUID = new ChannelGroupUID(thingUID, UIDUtils.encode(nodeID));
  77         properties = new ChildMap<>();
  78     }
  79
  80     /**
  81      * Parse node properties. This will not subscribe to properties though. Call
  82      * {@link Device#startChannels(MqttBrokerConnection, ScheduledExecutorService, int, HomieThingHandler)}
  83      * as soon as the returned future has completed.
  84      */
  85     public CompletableFuture<@Nullable Void> subscribe(MqttBrokerConnection connection,
  86             ScheduledExecutorService scheduler, int timeout) {
  87         return attributes.subscribeAndReceive(connection, scheduler, topic, this, timeout)
  88                 // On success, create all properties and tell the handler about this node
  89                 .thenCompose(b -> attributesReceived(connection, scheduler, timeout))
  90                 // No matter if values have been received or not -> the subscriptions have been performed
  91                 .whenComplete((r, e) -> {
  92                     initialized = true;
  93                 });
  94     }
  95
  96     public CompletableFuture<@Nullable Void> attributesReceived(MqttBrokerConnection connection,
  97             ScheduledExecutorService scheduler, int timeout) {
  98         callback.nodeAddedOrChanged(this);
  99         return applyProperties(connection, scheduler, timeout);
 100     }
 101
 102     public void nodeRestoredFromConfig() {
 103         initialized = true;
 104     }
 105
 106     /**
 107      * Unsubscribe from node attribute and also all property attributes and the property value
 108      *
 109      * @return Returns a future that completes as soon as all unsubscriptions have been performed.
 110      */
 111     public CompletableFuture<@Nullable Void> stop() {
 112         return attributes.unsubscribe().thenCompose(b -> CompletableFuture
 113                 .allOf(properties.stream().map(Property::stop).toArray(CompletableFuture[]::new)));
 114     }
 115
 116     /**
 117      * Return the channel group type for this Node.
 118      */
 119     public ChannelGroupType type() {
 120         final List<ChannelDefinition> channelDefinitions = properties.stream()
 121                 .map(c -> new ChannelDefinitionBuilder(c.propertyID, c.channelTypeUID).build())
 122                 .collect(Collectors.toList());
 123         return ChannelGroupTypeBuilder.instance(channelGroupTypeUID, attributes.name)
 124                 .withChannelDefinitions(channelDefinitions).build();
 125     }
 126
 127     /**
 128      * Return the channel group UID.
 129      */
 130     public ChannelGroupUID uid() {
 131         return channelGroupUID;
 132     }
 133
 134     /**
 135      * Create a Homie Property for this Node.
 136      *
 137      * @param propertyID The property ID
 138      * @return A Homie Property
 139      */
 140     public Property createProperty(String propertyID) {
 141         return new Property(topic, this, propertyID, callback, new PropertyAttributes());
 142     }
 143
 144     /**
 145      * Create a Homie Property for this Node.
 146      *
 147      * @param propertyID The property ID
 148      * @param attributes The node attributes object
 149      * @return A Homie Property
 150      */
 151     public Property createProperty(String propertyID, PropertyAttributes attributes) {
 152         return new Property(topic, this, propertyID, callback, attributes);
 153     }
 154
 155     /**
 156      * <p>
 157      * The properties of a node are determined by the node attribute "$properties". If that attribute changes,
 158      * {@link #attributeChanged(String, Object, MqttBrokerConnection, ScheduledExecutorServic, boolean)} is
 159      * called. The {@link #properties} map will be synchronized and this method will be called for every removed
 160      * property.
 161      * </p>
 162      *
 163      * <p>
 164      * This method will stop the property and will notify about the removed property.
 165      * </p>
 166      *
 167      * @param property The removed property.
 168      */
 169     protected void notifyPropertyRemoved(Property property) {
 170         property.stop();
 171         callback.propertyRemoved(property);
 172     }
 173
 174     protected CompletableFuture<@Nullable Void> applyProperties(MqttBrokerConnection connection,
 175             ScheduledExecutorService scheduler, int timeout) {
 176         return properties.apply(Objects.requireNonNull(attributes.properties),
 177                 prop -> prop.subscribe(connection, scheduler, timeout), this::createProperty,
 178                 this::notifyPropertyRemoved).exceptionally(e -> {
 179                     logger.warn("Could not subscribe", e);
 180                     return null;
 181                 });
 182     }
 183
 184     @Override
 185     public void attributeChanged(String name, Object value, MqttBrokerConnection connection,
 186             ScheduledExecutorService scheduler, boolean allMandatoryFieldsReceived) {
 187         if (!initialized || !allMandatoryFieldsReceived) {
 188             return;
 189         }
 190         // Special case: Not all fields were known before
 191         if (!attributes.isComplete()) {
 192             attributesReceived(connection, scheduler, 500);
 193         } else if ("properties".equals(name)) {
 194             applyProperties(connection, scheduler, 500);
 195         }
 196         callback.nodeAddedOrChanged(this);
 197     }
 198
 199     @Override
 200     public String toString() {
 201         return channelGroupUID.toString();
 202     }
 203
 204     /**
 205      * Creates a list of retained topics related to the node
 206      *
 207      * @return Returns a list of relative topics
 208      */
 209     public List<String> getRetainedTopics() {
 210         List<String> topics = new ArrayList<>(Stream.of(this.attributes.getClass().getDeclaredFields())
 211                 .map(f -> String.format("%s/$%s", this.nodeID, f.getName())).collect(Collectors.toList()));
 212
 213         this.properties.stream().map(p -> p.getRetainedTopics().stream()
 214                 .map(a -> String.format("%s/%s", this.nodeID, a)).collect(Collectors.toList()))
 215                 .collect(Collectors.toList()).forEach(topics::addAll);
 216
 217         return topics;
 218     }
 219 }