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

   1 /**
   2  * Copyright (c) 2010-2022 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.concurrent.CompletableFuture;
  18 import java.util.concurrent.ScheduledExecutorService;
  19 import java.util.stream.Collectors;
  20 import java.util.stream.Stream;
  21
  22 import org.eclipse.jdt.annotation.NonNullByDefault;
  23 import org.eclipse.jdt.annotation.Nullable;
  24 import org.openhab.binding.mqtt.generic.mapping.AbstractMqttAttributeClass;
  25 import org.openhab.binding.mqtt.generic.tools.ChildMap;
  26 import org.openhab.binding.mqtt.homie.generic.internal.MqttBindingConstants;
  27 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
  28 import org.openhab.core.thing.ChannelGroupUID;
  29 import org.openhab.core.thing.ThingUID;
  30 import org.openhab.core.thing.type.ChannelDefinition;
  31 import org.openhab.core.thing.type.ChannelDefinitionBuilder;
  32 import org.openhab.core.thing.type.ChannelGroupType;
  33 import org.openhab.core.thing.type.ChannelGroupTypeBuilder;
  34 import org.openhab.core.thing.type.ChannelGroupTypeUID;
  35 import org.openhab.core.util.UIDUtils;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 /**
  40  * Homie 3.x Node.
  41  *
  42  * A Homie Node contains Homie Properties ({@link Property}) but can also have attributes ({@link NodeAttributes}).
  43  * It corresponds to a ChannelGroup.
  44  *
  45  * @author David Graeff - Initial contribution
  46  */
  47 @NonNullByDefault
  48 public class Node implements AbstractMqttAttributeClass.AttributeChanged {
  49     private final Logger logger = LoggerFactory.getLogger(Node.class);
  50     // Homie
  51     public final String nodeID;
  52     public final NodeAttributes attributes;
  53     public ChildMap<Property> properties;
  54     // Runtime
  55     public final DeviceCallback callback;
  56     protected final ChannelGroupUID channelGroupUID;
  57     public final ChannelGroupTypeUID channelGroupTypeUID;
  58     private final String topic;
  59     private boolean initialized = false;
  60
  61     /**
  62      * Creates a Homie Node.
  63      *
  64      * @param topic The base topic for this node (e.g. "homie/device")
  65      * @param nodeID The node ID
  66      * @param thingUID The Thing UID, used to determine the ChannelGroupUID.
  67      * @param callback The callback for the handler.
  68      */
  69     public Node(String topic, String nodeID, ThingUID thingUID, DeviceCallback callback, NodeAttributes attributes) {
  70         this.attributes = attributes;
  71         this.topic = topic + "/" + nodeID;
  72         this.nodeID = nodeID;
  73         this.callback = callback;
  74         channelGroupTypeUID = new ChannelGroupTypeUID(MqttBindingConstants.BINDING_ID, UIDUtils.encode(this.topic));
  75         channelGroupUID = new ChannelGroupUID(thingUID, UIDUtils.encode(nodeID));
  76         properties = new ChildMap<>();
  77     }
  78
  79     /**
  80      * Parse node properties. This will not subscribe to properties though. Call
  81      * {@link Device#startChannels(MqttBrokerConnection)} as soon as the returned future has
  82      * completed.
  83      */
  84     public CompletableFuture<@Nullable Void> subscribe(MqttBrokerConnection connection,
  85             ScheduledExecutorService scheduler, int timeout) {
  86         return attributes.subscribeAndReceive(connection, scheduler, topic, this, timeout)
  87                 // On success, create all properties and tell the handler about this node
  88                 .thenCompose(b -> attributesReceived(connection, scheduler, timeout))
  89                 // No matter if values have been received or not -> the subscriptions have been performed
  90                 .whenComplete((r, e) -> {
  91                     initialized = true;
  92                 });
  93     }
  94
  95     public CompletableFuture<@Nullable Void> attributesReceived(MqttBrokerConnection connection,
  96             ScheduledExecutorService scheduler, int timeout) {
  97         callback.nodeAddedOrChanged(this);
  98         return applyProperties(connection, scheduler, timeout);
  99     }
 100
 101     public void nodeRestoredFromConfig() {
 102         initialized = true;
 103     }
 104
 105     /**
 106      * Unsubscribe from node attribute and also all property attributes and the property value
 107      *
 108      * @param connection A broker connection
 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(CompletableFuture, String, Object, MqttBrokerConnection, ScheduledExecutorService)} 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(attributes.properties, prop -> prop.subscribe(connection, scheduler, timeout),
 177                 this::createProperty, this::notifyPropertyRemoved).exceptionally(e -> {
 178                     logger.warn("Could not subscribe", e);
 179                     return null;
 180                 });
 181     }
 182
 183     @Override
 184     public void attributeChanged(String name, Object value, MqttBrokerConnection connection,
 185             ScheduledExecutorService scheduler, boolean allMandatoryFieldsReceived) {
 186         if (!initialized || !allMandatoryFieldsReceived) {
 187             return;
 188         }
 189         // Special case: Not all fields were known before
 190         if (!attributes.isComplete()) {
 191             attributesReceived(connection, scheduler, 500);
 192         } else {
 193             if ("properties".equals(name)) {
 194                 applyProperties(connection, scheduler, 500);
 195             }
 196         }
 197         callback.nodeAddedOrChanged(this);
 198     }
 199
 200     @Override
 201     public String toString() {
 202         return channelGroupUID.toString();
 203     }
 204
 205     /**
 206      * Creates a list of retained topics related to the node
 207      *
 208      * @return Returns a list of relative topics
 209      */
 210     public List<String> getRetainedTopics() {
 211         List<String> topics = new ArrayList<>();
 212
 213         topics.addAll(Stream.of(this.attributes.getClass().getDeclaredFields()).map(f -> {
 214             return String.format("%s/$%s", this.nodeID, f.getName());
 215         }).collect(Collectors.toList()));
 216
 217         this.properties.stream().map(p -> p.getRetainedTopics().stream().map(a -> {
 218             return String.format("%s/%s", this.nodeID, a);
 219         }).collect(Collectors.toList())).collect(Collectors.toList()).forEach(topics::addAll);
 220
 221         return topics;
 222     }
 223 }