Implemented rate limiting detection while posting events

Author: pc9795

src/main/java/com/exceptionless/exceptionlessclient/queue/DefaultEventQueue.java

@@ -4,9 +4,9 @@
 import com.exceptionless.exceptionlessclient.exceptions.SubmissionClientException;
 import com.exceptionless.exceptionlessclient.models.Event;
 import com.exceptionless.exceptionlessclient.models.storage.StorageItem;
-import com.exceptionless.exceptionlessclient.submission.SubmissionResponse;
 import com.exceptionless.exceptionlessclient.storage.StorageProviderIF;
 import com.exceptionless.exceptionlessclient.submission.SubmissionClientIF;
+import com.exceptionless.exceptionlessclient.submission.SubmissionResponse;
 import com.exceptionless.exceptionlessclient.utils.VisibleForTesting;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
@@ -158,6 +158,13 @@ private void processSubmissionResponse(
       return;
     }
 
+    if (response.isRateLimited()) {
+      log.error(
+          "Service is rate limited because of either you have exceeded your rate limit or server is under stress.");
+      suspendProcessing();
+      return;
+    }
+
     if (response.isServiceUnavailable()) {
       log.error("Service returns service unavailable");
       suspendProcessing();

src/main/java/com/exceptionless/exceptionlessclient/submission/DefaultSubmissionClient.java

@@ -15,12 +15,12 @@
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 @Slf4j
 public class DefaultSubmissionClient implements SubmissionClientIF {
   private static final String CONFIGURATION_VERSION_HEADER = "x-exceptionless-configversion";
+  private static final String RATE_LIMITING_HEADER = "x-ratelimit-remaining";
 
   private final Configuration configuration;
   private final SettingsManager settingsManager;
@@ -75,29 +75,33 @@ private SubmissionResponse postSubmission(String url, Object data) {
       Response response = httpClient.newCall(request).execute();
 
       if (response.code() / 100 == 2) {
-        updateSettingsFromHeaders(response.headers());
+        updateSettingsFromHeaders(response);
       }
 
       ResponseBody body = response.body();
       return SubmissionResponse.builder()
           .code(response.code())
           .body(body == null ? "" : body.string())
+          .rateLimitingHeaderFound(isRateLimitingHeaderFound(response))
           .build();
     } catch (Exception e) {
       throw new SubmissionClientException(e);
     }
   }
 
-  private void updateSettingsFromHeaders(Headers headers) {
-    Optional<String> maybeSettingsVersion =
-        Optional.ofNullable(headers.get(CONFIGURATION_VERSION_HEADER));
-    if (maybeSettingsVersion.isPresent()) {
-      settingsManager.checkVersion(Long.parseLong(maybeSettingsVersion.get()));
+  private void updateSettingsFromHeaders(Response response) {
+    String settingsVersion = response.headers().get(CONFIGURATION_VERSION_HEADER);
+    if (settingsVersion != null) {
+      settingsManager.checkVersion(Long.parseLong(settingsVersion));
     } else {
       log.error("No config version header was returned");
     }
   }
 
+  private boolean isRateLimitingHeaderFound(Response response) {
+    return response.headers().get(RATE_LIMITING_HEADER) != null;
+  }
+
   @Override
   public void sendHeartBeat(String sessionIdOrUserId, boolean closeSession) {
     try {

src/main/java/com/exceptionless/exceptionlessclient/submission/SubmissionResponse.java

@@ -8,6 +8,7 @@
 public class SubmissionResponse {
   int code;
   String body;
+  boolean rateLimitingHeaderFound;
 
   public boolean isSuccess() {
     return code >= 200 && code <= 299;
@@ -36,4 +37,8 @@ public boolean isNotFound() {
   public boolean isRequestEntityTooLarge() {
     return code == 413;
   }
+
+  public boolean isRateLimited() {
+    return rateLimitingHeaderFound || code == 429;
+  }
 }

src/test/java/com/exceptionless/exceptionlessclient/queue/DefaultEventQueueTest.java

@@ -4,10 +4,10 @@
 import com.exceptionless.exceptionlessclient.configuration.Configuration;
 import com.exceptionless.exceptionlessclient.exceptions.SubmissionClientException;
 import com.exceptionless.exceptionlessclient.models.Event;
-import com.exceptionless.exceptionlessclient.submission.SubmissionResponse;
 import com.exceptionless.exceptionlessclient.storage.InMemoryStorage;
 import com.exceptionless.exceptionlessclient.storage.InMemoryStorageProvider;
 import com.exceptionless.exceptionlessclient.submission.DefaultSubmissionClient;
+import com.exceptionless.exceptionlessclient.submission.SubmissionResponse;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -111,7 +111,9 @@ public void itShouldNotPostEmptyEvents() {
   public void itShouldSuspendProcessingOnClientException() {
     storage.save(event);
 
-    doThrow(new SubmissionClientException("test")).when(submissionClient).postEvents(List.of(event));
+    doThrow(new SubmissionClientException("test"))
+        .when(submissionClient)
+        .postEvents(List.of(event));
 
     queue.onEventsPosted(testHandler);
     queue.process();
@@ -137,6 +139,42 @@ public void itShouldSuspendProcessingIfServiceIsUnavailable() {
     verify(testHandler, times(1)).accept(List.of(event), response);
   }
 
+  @Test
+  public void itShouldSuspendProcessingIfServiceIsRateLimitedByHeader() {
+    storage.save(event);
+
+    SubmissionResponse response =
+        SubmissionResponse.builder()
+            .body("test-message")
+            .code(999)
+            .rateLimitingHeaderFound(true)
+            .build();
+    doReturn(response).when(submissionClient).postEvents(List.of(event));
+
+    queue.onEventsPosted(testHandler);
+    queue.process();
+
+    assertThat(queue.isProcessingCurrentlySuspended()).isTrue();
+    assertThat(storage.peek().getValue()).isEqualTo(event);
+    verify(testHandler, times(1)).accept(List.of(event), response);
+  }
+
+  @Test
+  public void itShouldSuspendProcessingIfServiceIsRateLimitedByCode() {
+    storage.save(event);
+
+    SubmissionResponse response =
+        SubmissionResponse.builder().body("test-message").code(429).build();
+    doReturn(response).when(submissionClient).postEvents(List.of(event));
+
+    queue.onEventsPosted(testHandler);
+    queue.process();
+
+    assertThat(queue.isProcessingCurrentlySuspended()).isTrue();
+    assertThat(storage.peek().getValue()).isEqualTo(event);
+    verify(testHandler, times(1)).accept(List.of(event), response);
+  }
+
   @Test
   public void itShouldSuspendAndDiscardProcessingAndClearQueueIfNoPayment() {
     storage.save(event);
@@ -250,11 +288,9 @@ public void itShouldReduceSubmissionBatchSizeIfRequestEntitiesAreTooLarge() {
     queue.process();
 
     // One invocation with full batch
-    verify(submissionClient, times(1))
-        .postEvents(argThat(argument -> argument.size() == 3));
+    verify(submissionClient, times(1)).postEvents(argThat(argument -> argument.size() == 3));
     // One invocation with reduced batch
-    verify(submissionClient, times(1))
-        .postEvents(argThat(argument -> argument.size() == 2));
+    verify(submissionClient, times(1)).postEvents(argThat(argument -> argument.size() == 2));
   }
 
   @Test
@@ -303,11 +339,9 @@ public void itShouldResetSubmissionBatchSizeOnNextSuccessfulResponse() {
 
     // Two invocations with full batch; First with the default size and next after a successful
     // response
-    verify(submissionClient, times(2))
-        .postEvents(argThat(argument -> argument.size() == 3));
+    verify(submissionClient, times(2)).postEvents(argThat(argument -> argument.size() == 3));
     // One invocation with reduced batch
-    verify(submissionClient, times(1))
-        .postEvents(argThat(argument -> argument.size() == 2));
+    verify(submissionClient, times(1)).postEvents(argThat(argument -> argument.size() == 2));
   }
 
   @Test

src/test/java/com/exceptionless/exceptionlessclient/submission/DefaultSubmissionClientTest.java

@@ -76,6 +76,36 @@ public void itCanPostEventsSuccessfully() throws IOException {
     verify(settingsManager, times(1)).checkVersion(3);
   }
 
+  @Test
+  public void itCanDetectRateLimitingFromHeaders() throws IOException {
+    Response response =
+        responseBuilder
+            .headers(
+                Headers.of(
+                    Map.of("x-exceptionless-configversion", "3", "x-ratelimit-remaining", "0")))
+            .build();
+    doReturn(response).when(call).execute();
+    doReturn(call)
+        .when(httpClient)
+        .newCall(
+            argThat(
+                request ->
+                    request
+                            .url()
+                            .toString()
+                            .equals(
+                                "http://test-server-url/api/v2/events?access_token=test-api-key")
+                        && request.method().equals("POST")));
+
+    SubmissionResponse submissionResponse =
+        submissionClient.postEvents(List.of(Event.builder().build()));
+
+    assertThat(submissionResponse.getBody()).isEqualTo("test-body");
+    assertThat(submissionResponse.getCode()).isEqualTo(200);
+    verify(settingsManager, times(1)).checkVersion(3);
+    assertThat(submissionResponse.isRateLimited()).isTrue();
+  }
+
   @Test
   public void itCanPostEventsSuccessfullyWhenNoSettingHeaderIsReturned() throws IOException {
     doReturn(responseBuilder.build()).when(call).execute();