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.http.internal.http;
  14
  15 import java.net.URI;
  16 import java.net.URISyntaxException;
  17 import java.util.Date;
  18 import java.util.List;
  19 import java.util.Optional;
  20 import java.util.Set;
  21 import java.util.concurrent.*;
  22 import java.util.function.Consumer;
  23
  24 import org.eclipse.jdt.annotation.NonNullByDefault;
  25 import org.eclipse.jdt.annotation.Nullable;
  26 import org.eclipse.jetty.client.api.Authentication;
  27 import org.eclipse.jetty.client.api.AuthenticationStore;
  28 import org.openhab.binding.http.internal.Util;
  29 import org.openhab.binding.http.internal.config.HttpThingConfig;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 /**
  34  * The {@link RefreshingUrlCache} is responsible for requesting from a single URL and passing the content to the
  35  * channels
  36  *
  37  * @author Jan N. Klug - Initial contribution
  38  */
  39 @NonNullByDefault
  40 public class RefreshingUrlCache {
  41     private final Logger logger = LoggerFactory.getLogger(RefreshingUrlCache.class);
  42
  43     private final String url;
  44     private final RateLimitedHttpClient httpClient;
  45     private final int timeout;
  46     private final int bufferSize;
  47     private final @Nullable String fallbackEncoding;
  48     private final Set<Consumer<Content>> consumers = ConcurrentHashMap.newKeySet();
  49     private final List<String> headers;
  50
  51     private final ScheduledFuture<?> future;
  52     private @Nullable Content lastContent;
  53
  54     public RefreshingUrlCache(ScheduledExecutorService executor, RateLimitedHttpClient httpClient, String url,
  55             HttpThingConfig thingConfig) {
  56         this.httpClient = httpClient;
  57         this.url = url;
  58         this.timeout = thingConfig.timeout;
  59         this.bufferSize = thingConfig.bufferSize;
  60         this.headers = thingConfig.headers;
  61         fallbackEncoding = thingConfig.encoding;
  62
  63         future = executor.scheduleWithFixedDelay(this::refresh, 0, thingConfig.refresh, TimeUnit.SECONDS);
  64         logger.trace("Started refresh task for URL '{}' with interval {}s", url, thingConfig.refresh);
  65     }
  66
  67     private void refresh() {
  68         refresh(false);
  69     }
  70
  71     private void refresh(boolean isRetry) {
  72         if (consumers.isEmpty()) {
  73             // do not refresh if we don't have listeners
  74             return;
  75         }
  76
  77         // format URL
  78         try {
  79             URI finalUrl = new URI(String.format(this.url, new Date()));
  80
  81             logger.trace("Requesting refresh (retry={}) from '{}' with timeout {}ms", isRetry, finalUrl, timeout);
  82
  83             httpClient.newRequest(finalUrl).thenAccept(request -> {
  84                 request.timeout(timeout, TimeUnit.MILLISECONDS);
  85
  86                 headers.forEach(header -> {
  87                     String[] keyValuePair = header.split("=", 2);
  88                     if (keyValuePair.length == 2) {
  89                         request.header(keyValuePair[0].trim(), keyValuePair[1].trim());
  90                     } else {
  91                         logger.warn("Splitting header '{}' failed. No '=' was found. Ignoring", header);
  92                     }
  93                 });
  94
  95                 CompletableFuture<@Nullable Content> response = new CompletableFuture<>();
  96                 response.exceptionally(e -> {
  97                     if (e instanceof HttpAuthException) {
  98                         if (isRetry) {
  99                             logger.warn("Retry after authentication  failure failed again for '{}', failing here",
 100                                     finalUrl);
 101                         } else {
 102                             AuthenticationStore authStore = httpClient.getAuthenticationStore();
 103                             Authentication.Result authResult = authStore.findAuthenticationResult(finalUrl);
 104                             if (authResult != null) {
 105                                 authStore.removeAuthenticationResult(authResult);
 106                                 logger.debug("Cleared authentication result for '{}', retrying immediately", finalUrl);
 107                                 refresh(true);
 108                             } else {
 109                                 logger.warn("Could not find authentication result for '{}', failing here", finalUrl);
 110                             }
 111                         }
 112                     }
 113                     return null;
 114                 }).thenAccept(this::processResult);
 115
 116                 if (logger.isTraceEnabled()) {
 117                     logger.trace("Sending to '{}': {}", finalUrl, Util.requestToLogString(request));
 118                 }
 119
 120                 request.send(new HttpResponseListener(response, fallbackEncoding, bufferSize));
 121             }).exceptionally(e -> {
 122                 if (e instanceof CancellationException) {
 123                     logger.debug("Request to URL {} was cancelled by thing handler.", finalUrl);
 124                 } else {
 125                     logger.warn("Request to URL {} failed: {}", finalUrl, e.getMessage());
 126                 }
 127                 return null;
 128             });
 129         } catch (IllegalArgumentException | URISyntaxException e) {
 130             logger.warn("Creating request for '{}' failed: {}", url, e.getMessage());
 131         }
 132     }
 133
 134     public void stop() {
 135         // clearing all listeners to prevent further updates
 136         consumers.clear();
 137         future.cancel(false);
 138         logger.trace("Stopped refresh task for URL '{}'", url);
 139     }
 140
 141     public void addConsumer(Consumer<Content> consumer) {
 142         consumers.add(consumer);
 143     }
 144
 145     public Optional<Content> get() {
 146         final Content content = lastContent;
 147         if (content == null) {
 148             return Optional.empty();
 149         } else {
 150             return Optional.of(content);
 151         }
 152     }
 153
 154     private void processResult(@Nullable Content content) {
 155         if (content != null) {
 156             for (Consumer<Content> consumer : consumers) {
 157                 try {
 158                     consumer.accept(content);
 159                 } catch (IllegalArgumentException | IllegalStateException e) {
 160                     logger.warn("Failed processing result for URL {}: {}", url, e.getMessage());
 161                 }
 162             }
 163         }
 164         lastContent = content;
 165     }
 166 }