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