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