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

   1 /**
   2  * Copyright (c) 2010-2020 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.knx.internal.client;
  14
  15 import java.util.Set;
  16 import java.util.concurrent.CopyOnWriteArraySet;
  17 import java.util.concurrent.LinkedBlockingQueue;
  18 import java.util.concurrent.ScheduledExecutorService;
  19 import java.util.concurrent.ScheduledFuture;
  20 import java.util.concurrent.TimeUnit;
  21 import java.util.function.Consumer;
  22
  23 import org.eclipse.jdt.annotation.NonNullByDefault;
  24 import org.eclipse.jdt.annotation.Nullable;
  25 import org.openhab.binding.knx.internal.KNXTypeMapper;
  26 import org.openhab.binding.knx.internal.dpt.KNXCoreTypeMapper;
  27 import org.openhab.binding.knx.internal.handler.GroupAddressListener;
  28 import org.openhab.core.thing.ThingStatus;
  29 import org.openhab.core.thing.ThingStatusDetail;
  30 import org.openhab.core.thing.ThingUID;
  31 import org.openhab.core.types.Type;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 import tuwien.auto.calimero.CloseEvent;
  36 import tuwien.auto.calimero.DetachEvent;
  37 import tuwien.auto.calimero.FrameEvent;
  38 import tuwien.auto.calimero.GroupAddress;
  39 import tuwien.auto.calimero.IndividualAddress;
  40 import tuwien.auto.calimero.KNXException;
  41 import tuwien.auto.calimero.datapoint.CommandDP;
  42 import tuwien.auto.calimero.datapoint.Datapoint;
  43 import tuwien.auto.calimero.device.ProcessCommunicationResponder;
  44 import tuwien.auto.calimero.link.KNXNetworkLink;
  45 import tuwien.auto.calimero.link.NetworkLinkListener;
  46 import tuwien.auto.calimero.mgmt.Destination;
  47 import tuwien.auto.calimero.mgmt.ManagementClient;
  48 import tuwien.auto.calimero.mgmt.ManagementClientImpl;
  49 import tuwien.auto.calimero.mgmt.ManagementProcedures;
  50 import tuwien.auto.calimero.mgmt.ManagementProceduresImpl;
  51 import tuwien.auto.calimero.process.ProcessCommunicationBase;
  52 import tuwien.auto.calimero.process.ProcessCommunicator;
  53 import tuwien.auto.calimero.process.ProcessCommunicatorImpl;
  54 import tuwien.auto.calimero.process.ProcessEvent;
  55 import tuwien.auto.calimero.process.ProcessListener;
  56
  57 /**
  58  * KNX Client which encapsulates the communication with the KNX bus via the calimero libary.
  59  *
  60  * @author Simon Kaufmann - initial contribution and API.
  61  *
  62  */
  63 @NonNullByDefault
  64 public abstract class AbstractKNXClient implements NetworkLinkListener, KNXClient {
  65
  66     private static final int MAX_SEND_ATTEMPTS = 2;
  67
  68     private final Logger logger = LoggerFactory.getLogger(AbstractKNXClient.class);
  69     private final KNXTypeMapper typeHelper = new KNXCoreTypeMapper();
  70
  71     private final ThingUID thingUID;
  72     private final int responseTimeout;
  73     private final int readingPause;
  74     private final int autoReconnectPeriod;
  75     private final int readRetriesLimit;
  76     private final StatusUpdateCallback statusUpdateCallback;
  77     private final ScheduledExecutorService knxScheduler;
  78
  79     private @Nullable ProcessCommunicator processCommunicator;
  80     private @Nullable ProcessCommunicationResponder responseCommunicator;
  81     private @Nullable ManagementProcedures managementProcedures;
  82     private @Nullable ManagementClient managementClient;
  83     private @Nullable KNXNetworkLink link;
  84     private @Nullable DeviceInfoClient deviceInfoClient;
  85     private @Nullable ScheduledFuture<?> busJob;
  86     private @Nullable ScheduledFuture<?> connectJob;
  87
  88     private final Set<GroupAddressListener> groupAddressListeners = new CopyOnWriteArraySet<>();
  89     private final LinkedBlockingQueue<ReadDatapoint> readDatapoints = new LinkedBlockingQueue<>();
  90
  91     @FunctionalInterface
  92     private interface ListenerNotification {
  93         void apply(BusMessageListener listener, IndividualAddress source, GroupAddress destination, byte[] asdu);
  94     }
  95
  96     @NonNullByDefault({})
  97     private final ProcessListener processListener = new ProcessListener() {
  98
  99         @Override
 100         public void detached(DetachEvent e) {
 101             logger.debug("The KNX network link was detached from the process communicator");
 102         }
 103
 104         @Override
 105         public void groupWrite(ProcessEvent e) {
 106             processEvent("Group Write", e, (listener, source, destination, asdu) -> {
 107                 listener.onGroupWrite(AbstractKNXClient.this, source, destination, asdu);
 108             });
 109         }
 110
 111         @Override
 112         public void groupReadRequest(ProcessEvent e) {
 113             processEvent("Group Read Request", e, (listener, source, destination, asdu) -> {
 114                 listener.onGroupRead(AbstractKNXClient.this, source, destination, asdu);
 115             });
 116         }
 117
 118         @Override
 119         public void groupReadResponse(ProcessEvent e) {
 120             processEvent("Group Read Response", e, (listener, source, destination, asdu) -> {
 121                 listener.onGroupReadResponse(AbstractKNXClient.this, source, destination, asdu);
 122             });
 123         }
 124     };
 125
 126     public AbstractKNXClient(int autoReconnectPeriod, ThingUID thingUID, int responseTimeout, int readingPause,
 127             int readRetriesLimit, ScheduledExecutorService knxScheduler, StatusUpdateCallback statusUpdateCallback) {
 128         this.autoReconnectPeriod = autoReconnectPeriod;
 129         this.thingUID = thingUID;
 130         this.responseTimeout = responseTimeout;
 131         this.readingPause = readingPause;
 132         this.readRetriesLimit = readRetriesLimit;
 133         this.knxScheduler = knxScheduler;
 134         this.statusUpdateCallback = statusUpdateCallback;
 135     }
 136
 137     public void initialize() {
 138         if (!scheduleReconnectJob()) {
 139             connect();
 140         }
 141     }
 142
 143     private boolean scheduleReconnectJob() {
 144         if (autoReconnectPeriod > 0) {
 145             connectJob = knxScheduler.schedule(this::connect, autoReconnectPeriod, TimeUnit.SECONDS);
 146             return true;
 147         } else {
 148             return false;
 149         }
 150     }
 151
 152     private void cancelReconnectJob() {
 153         ScheduledFuture<?> currentReconnectJob = connectJob;
 154         if (currentReconnectJob != null) {
 155             currentReconnectJob.cancel(true);
 156             connectJob = null;
 157         }
 158     }
 159
 160     protected abstract KNXNetworkLink establishConnection() throws KNXException, InterruptedException;
 161
 162     private synchronized boolean connectIfNotAutomatic() {
 163         if (!isConnected()) {
 164             return connectJob != null ? false : connect();
 165         }
 166         return true;
 167     }
 168
 169     private synchronized boolean connect() {
 170         if (isConnected()) {
 171             return true;
 172         }
 173         try {
 174             releaseConnection();
 175
 176             logger.debug("Bridge {} is connecting to the KNX bus", thingUID);
 177
 178             KNXNetworkLink link = establishConnection();
 179             this.link = link;
 180
 181             managementProcedures = new ManagementProceduresImpl(link);
 182
 183             ManagementClient managementClient = new ManagementClientImpl(link);
 184             managementClient.setResponseTimeout(responseTimeout);
 185             this.managementClient = managementClient;
 186
 187             deviceInfoClient = new DeviceInfoClientImpl(managementClient);
 188
 189             ProcessCommunicator processCommunicator = new ProcessCommunicatorImpl(link);
 190             processCommunicator.setResponseTimeout(responseTimeout);
 191             processCommunicator.addProcessListener(processListener);
 192             this.processCommunicator = processCommunicator;
 193
 194             ProcessCommunicationResponder responseCommunicator = new ProcessCommunicationResponder(link);
 195             this.responseCommunicator = responseCommunicator;
 196
 197             link.addLinkListener(this);
 198
 199             busJob = knxScheduler.scheduleWithFixedDelay(() -> readNextQueuedDatapoint(), 0, readingPause,
 200                     TimeUnit.MILLISECONDS);
 201
 202             statusUpdateCallback.updateStatus(ThingStatus.ONLINE);
 203             connectJob = null;
 204             return true;
 205         } catch (KNXException | InterruptedException e) {
 206             logger.debug("Error connecting to the bus: {}", e.getMessage(), e);
 207             disconnect(e);
 208             scheduleReconnectJob();
 209             return false;
 210         }
 211     }
 212
 213     private void disconnect(@Nullable Exception e) {
 214         releaseConnection();
 215         if (e != null) {
 216             statusUpdateCallback.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
 217                     e.getLocalizedMessage());
 218         } else {
 219             statusUpdateCallback.updateStatus(ThingStatus.OFFLINE);
 220         }
 221     }
 222
 223     @SuppressWarnings("null")
 224     private void releaseConnection() {
 225         logger.debug("Bridge {} is disconnecting from the KNX bus", thingUID);
 226         readDatapoints.clear();
 227         busJob = nullify(busJob, j -> j.cancel(true));
 228         deviceInfoClient = null;
 229         managementProcedures = nullify(managementProcedures, mp -> mp.detach());
 230         managementClient = nullify(managementClient, mc -> mc.detach());
 231         link = nullify(link, l -> l.close());
 232         processCommunicator = nullify(processCommunicator, pc -> {
 233             pc.removeProcessListener(processListener);
 234             pc.detach();
 235         });
 236         responseCommunicator = nullify(responseCommunicator, rc -> {
 237             rc.removeProcessListener(processListener);
 238             rc.detach();
 239         });
 240     }
 241
 242     private <T> T nullify(T target, @Nullable Consumer<T> lastWill) {
 243         if (target != null && lastWill != null) {
 244             lastWill.accept(target);
 245         }
 246         return null;
 247     }
 248
 249     private void processEvent(String task, ProcessEvent event, ListenerNotification action) {
 250         GroupAddress destination = event.getDestination();
 251         IndividualAddress source = event.getSourceAddr();
 252         byte[] asdu = event.getASDU();
 253         logger.trace("Received a {} telegram from '{}' to '{}' with value '{}'", task, source, destination, asdu);
 254         for (GroupAddressListener listener : groupAddressListeners) {
 255             if (listener.listensTo(destination)) {
 256                 knxScheduler.schedule(() -> action.apply(listener, source, destination, asdu), 0, TimeUnit.SECONDS);
 257             }
 258         }
 259     }
 260
 261     /**
 262      * Transforms a {@link Type} into a datapoint type value for the KNX bus.
 263      *
 264      * @param type the {@link Type} to transform
 265      * @param dpt the datapoint type to which should be converted
 266      * @return the corresponding KNX datapoint type value as a string
 267      */
 268     @Nullable
 269     private String toDPTValue(Type type, String dpt) {
 270         return typeHelper.toDPTValue(type, dpt);
 271     }
 272
 273     @SuppressWarnings("null")
 274     private void readNextQueuedDatapoint() {
 275         if (!connectIfNotAutomatic()) {
 276             return;
 277         }
 278         ProcessCommunicator processCommunicator = this.processCommunicator;
 279         if (processCommunicator == null) {
 280             return;
 281         }
 282         ReadDatapoint datapoint = readDatapoints.poll();
 283         if (datapoint != null) {
 284             datapoint.incrementRetries();
 285             try {
 286                 logger.trace("Sending a Group Read Request telegram for {}", datapoint.getDatapoint().getMainAddress());
 287                 processCommunicator.read(datapoint.getDatapoint());
 288             } catch (KNXException e) {
 289                 if (datapoint.getRetries() < datapoint.getLimit()) {
 290                     readDatapoints.add(datapoint);
 291                     logger.debug("Could not read value for datapoint {}: {}. Going to retry.",
 292                             datapoint.getDatapoint().getMainAddress(), e.getMessage());
 293                 } else {
 294                     logger.warn("Giving up reading datapoint {}, the number of maximum retries ({}) is reached.",
 295                             datapoint.getDatapoint().getMainAddress(), datapoint.getLimit());
 296                 }
 297             } catch (InterruptedException e) {
 298                 logger.debug("Interrupted sending KNX read request");
 299                 return;
 300             }
 301         }
 302     }
 303
 304     public void dispose() {
 305         cancelReconnectJob();
 306         disconnect(null);
 307     }
 308
 309     @Override
 310     public void linkClosed(@Nullable CloseEvent closeEvent) {
 311         KNXNetworkLink link = this.link;
 312         if (link == null || closeEvent == null) {
 313             return;
 314         }
 315         if (!link.isOpen() && CloseEvent.USER_REQUEST != closeEvent.getInitiator()) {
 316             statusUpdateCallback.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
 317                     closeEvent.getReason());
 318             logger.debug("KNX link has been lost (reason: {} on object {})", closeEvent.getReason(),
 319                     closeEvent.getSource().toString());
 320             scheduleReconnectJob();
 321         }
 322     }
 323
 324     @Override
 325     public void indication(@Nullable FrameEvent e) {
 326         // no-op
 327     }
 328
 329     @Override
 330     public void confirmation(@Nullable FrameEvent e) {
 331         // no-op
 332     }
 333
 334     @Override
 335     public final synchronized boolean isReachable(@Nullable IndividualAddress address) throws KNXException {
 336         ManagementProcedures managementProcedures = this.managementProcedures;
 337         if (managementProcedures == null || address == null) {
 338             return false;
 339         }
 340         try {
 341             return managementProcedures.isAddressOccupied(address);
 342         } catch (InterruptedException e) {
 343             logger.debug("Interrupted pinging KNX device '{}'", address);
 344         }
 345         return false;
 346     }
 347
 348     @Override
 349     public final synchronized void restartNetworkDevice(@Nullable IndividualAddress address) {
 350         ManagementClient managementClient = this.managementClient;
 351         if (address == null || managementClient == null) {
 352             return;
 353         }
 354         Destination destination = null;
 355         try {
 356             destination = managementClient.createDestination(address, true);
 357             managementClient.restart(destination);
 358         } catch (KNXException e) {
 359             logger.warn("Could not reset device with address '{}': {}", address, e.getMessage());
 360         } catch (InterruptedException e) { // ignored as in Calimero pre-2.4.0
 361         } finally {
 362             if (destination != null) {
 363                 destination.destroy();
 364             }
 365         }
 366     }
 367
 368     @Override
 369     public void readDatapoint(Datapoint datapoint) {
 370         synchronized (this) {
 371             ReadDatapoint retryDatapoint = new ReadDatapoint(datapoint, readRetriesLimit);
 372             if (!readDatapoints.contains(retryDatapoint)) {
 373                 readDatapoints.add(retryDatapoint);
 374             }
 375         }
 376     }
 377
 378     @Override
 379     public final boolean registerGroupAddressListener(GroupAddressListener listener) {
 380         return groupAddressListeners.add(listener);
 381     }
 382
 383     @Override
 384     public final boolean unregisterGroupAddressListener(GroupAddressListener listener) {
 385         return groupAddressListeners.remove(listener);
 386     }
 387
 388     @Override
 389     public boolean isConnected() {
 390         return link != null && link.isOpen();
 391     }
 392
 393     @Override
 394     public DeviceInfoClient getDeviceInfoClient() {
 395         DeviceInfoClient deviceInfoClient = this.deviceInfoClient;
 396         if (deviceInfoClient != null) {
 397             return deviceInfoClient;
 398         } else {
 399             throw new IllegalStateException();
 400         }
 401     }
 402
 403     @Override
 404     public void writeToKNX(OutboundSpec commandSpec) throws KNXException {
 405         ProcessCommunicator processCommunicator = this.processCommunicator;
 406         KNXNetworkLink link = this.link;
 407         if (processCommunicator == null || link == null) {
 408             logger.debug("Cannot write to the KNX bus (processCommuicator: {}, link: {})",
 409                     processCommunicator == null ? "Not OK" : "OK",
 410                     link == null ? "Not OK" : (link.isOpen() ? "Open" : "Closed"));
 411             return;
 412         }
 413         GroupAddress groupAddress = commandSpec.getGroupAddress();
 414
 415         logger.trace("writeToKNX groupAddress '{}', commandSpec '{}'", groupAddress, commandSpec);
 416
 417         if (groupAddress != null) {
 418             sendToKNX(processCommunicator, link, groupAddress, commandSpec.getDPT(), commandSpec.getType());
 419         }
 420     }
 421
 422     @Override
 423     public void respondToKNX(OutboundSpec responseSpec) throws KNXException {
 424         ProcessCommunicationResponder responseCommunicator = this.responseCommunicator;
 425         KNXNetworkLink link = this.link;
 426         if (responseCommunicator == null || link == null) {
 427             logger.debug("Cannot write to the KNX bus (responseCommunicator: {}, link: {})",
 428                     responseCommunicator == null ? "Not OK" : "OK",
 429                     link == null ? "Not OK" : (link.isOpen() ? "Open" : "Closed"));
 430             return;
 431         }
 432         GroupAddress groupAddress = responseSpec.getGroupAddress();
 433
 434         logger.trace("respondToKNX groupAddress '{}', responseSpec '{}'", groupAddress, responseSpec);
 435
 436         if (groupAddress != null) {
 437             sendToKNX(responseCommunicator, link, groupAddress, responseSpec.getDPT(), responseSpec.getType());
 438         }
 439     }
 440
 441     private void sendToKNX(ProcessCommunicationBase communicator, KNXNetworkLink link, GroupAddress groupAddress,
 442             String dpt, Type type) throws KNXException {
 443         if (!connectIfNotAutomatic()) {
 444             return;
 445         }
 446
 447         Datapoint datapoint = new CommandDP(groupAddress, thingUID.toString(), 0, dpt);
 448         String mappedValue = toDPTValue(type, dpt);
 449
 450         logger.trace("sendToKNX mappedValue: '{}' groupAddress: '{}'", mappedValue, groupAddress);
 451
 452         if (mappedValue == null) {
 453             logger.debug("Value '{}' cannot be mapped to datapoint '{}'", type, datapoint);
 454             return;
 455         }
 456         for (int i = 0; i < MAX_SEND_ATTEMPTS; i++) {
 457             try {
 458                 communicator.write(datapoint, mappedValue);
 459                 logger.debug("Wrote value '{}' to datapoint '{}' ({}. attempt).", type, datapoint, i);
 460                 break;
 461             } catch (KNXException e) {
 462                 if (i < MAX_SEND_ATTEMPTS - 1) {
 463                     logger.debug("Value '{}' could not be sent to the KNX bus using datapoint '{}': {}. Will retry.",
 464                             type, datapoint, e.getLocalizedMessage());
 465                 } else {
 466                     logger.warn("Value '{}' could not be sent to the KNX bus using datapoint '{}': {}. Giving up now.",
 467                             type, datapoint, e.getLocalizedMessage());
 468                     throw e;
 469                 }
 470             }
 471         }
 472     }
 473 }