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

   1 /**
   2  * Copyright (c) 2010-2021 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.util.concurrent.*;
  17
  18 import org.eclipse.jdt.annotation.NonNullByDefault;
  19 import org.eclipse.jdt.annotation.Nullable;
  20 import org.eclipse.jetty.client.HttpClient;
  21 import org.eclipse.jetty.client.api.AuthenticationStore;
  22 import org.eclipse.jetty.client.api.Request;
  23
  24 /**
  25  * The {@link RateLimitedHttpClient} is a wrapper for a Jetty HTTP client that limits the number of requests by delaying
  26  * the request creation
  27  *
  28  * @author Jan N. Klug - Initial contribution
  29  */
  30 @NonNullByDefault
  31 public class RateLimitedHttpClient {
  32     private static final int MAX_QUEUE_SIZE = 1000; // maximum queue size
  33     private HttpClient httpClient;
  34     private int delay = 0; // in ms
  35     private final ScheduledExecutorService scheduler;
  36     private final LinkedBlockingQueue<RequestQueueEntry> requestQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE);
  37
  38     private @Nullable ScheduledFuture<?> processJob;
  39
  40     public RateLimitedHttpClient(HttpClient httpClient, ScheduledExecutorService scheduler) {
  41         this.httpClient = httpClient;
  42         this.scheduler = scheduler;
  43     }
  44
  45     /**
  46      * Stop processing the queue and clear it
  47      */
  48     public void shutdown() {
  49         stopProcessJob();
  50         requestQueue.forEach(queueEntry -> queueEntry.future.completeExceptionally(new CancellationException()));
  51     }
  52
  53     /**
  54      * Set a new delay
  55      *
  56      * @param delay in ms between to requests
  57      */
  58     public void setDelay(int delay) {
  59         if (delay < 0) {
  60             throw new IllegalArgumentException("Delay needs to be larger or equal to zero");
  61         }
  62         this.delay = delay;
  63         stopProcessJob();
  64         if (delay != 0) {
  65             processJob = scheduler.scheduleWithFixedDelay(this::processQueue, 0, delay, TimeUnit.MILLISECONDS);
  66         }
  67     }
  68
  69     /**
  70      * Set the HTTP client
  71      *
  72      * @param httpClient secure or insecure Jetty http client
  73      */
  74     public void setHttpClient(HttpClient httpClient) {
  75         this.httpClient = httpClient;
  76     }
  77
  78     /**
  79      * Create a new request to the given URL respecting rate-limits
  80      *
  81      * @param finalUrl the request URL
  82      * @return a CompletableFuture that completes with the request
  83      */
  84     public CompletableFuture<Request> newRequest(URI finalUrl) {
  85         // if no delay is set, return a completed CompletableFuture
  86         if (delay == 0) {
  87             return CompletableFuture.completedFuture(httpClient.newRequest(finalUrl));
  88         }
  89         CompletableFuture<Request> future = new CompletableFuture<>();
  90         if (!requestQueue.offer(new RequestQueueEntry(finalUrl, future))) {
  91             future.completeExceptionally(new RejectedExecutionException("Maximum queue size exceeded."));
  92         }
  93         return future;
  94     }
  95
  96     /**
  97      * Get the AuthenticationStore from the wrapped client
  98      *
  99      * @return
 100      */
 101     public AuthenticationStore getAuthenticationStore() {
 102         return httpClient.getAuthenticationStore();
 103     }
 104
 105     private void stopProcessJob() {
 106         ScheduledFuture<?> processJob = this.processJob;
 107         if (processJob != null) {
 108             processJob.cancel(false);
 109             this.processJob = null;
 110         }
 111     }
 112
 113     private void processQueue() {
 114         RequestQueueEntry queueEntry = requestQueue.poll();
 115         if (queueEntry != null) {
 116             queueEntry.future.complete(httpClient.newRequest(queueEntry.finalUrl));
 117         }
 118     }
 119
 120     private static class RequestQueueEntry {
 121         public URI finalUrl;
 122         public CompletableFuture<Request> future;
 123
 124         public RequestQueueEntry(URI finalUrl, CompletableFuture<Request> future) {
 125             this.finalUrl = finalUrl;
 126             this.future = future;
 127         }
 128     }
 129 }