RPC headers: add request retries support

Author: mostroverkhov

rsocket-messages/src/main/java/com/jauntsdn/rsocket/Headers.java

@@ -34,19 +34,26 @@
 public final class Headers {
   public static int HEADER_LENGTH_MAX = 8192;
 
-  private static final Headers EMPTY = new Headers(false, 0, Collections.emptyList(), 0);
-  private static final Headers DEFAULT_SERVICE = new Headers(true, 0, Collections.emptyList(), 0);
+  private static final Headers EMPTY = new Headers(false, 0, null, Collections.emptyList(), 0);
+  private static final Headers DEFAULT_SERVICE =
+      new Headers(true, 0, null, Collections.emptyList(), 0);
 
   private final boolean isDefaultService;
   private final int serializedSize;
   private final long timeoutMillis;
+  private final RetryPolicy retryPolicy;
   private final List<String> keyValues;
   private volatile ByteBuf cache;
 
   private Headers(
-      boolean isDefaultService, long timeoutMillis, List<String> keyValues, int serializedSize) {
+      boolean isDefaultService,
+      long timeoutMillis,
+      @Nullable RetryPolicy retryPolicy,
+      List<String> keyValues,
+      int serializedSize) {
     this.isDefaultService = isDefaultService;
     this.timeoutMillis = timeoutMillis;
+    this.retryPolicy = retryPolicy;
     this.keyValues = keyValues;
     this.serializedSize = serializedSize;
   }
@@ -59,6 +66,10 @@ public long timeoutMillis() {
     return timeoutMillis;
   }
 
+  public RetryPolicy retryPolicy() {
+    return retryPolicy;
+  }
+
   public String header(String name) {
     if (!isValidKeySize(name)) {
       return null;
@@ -142,7 +153,16 @@ public String setValue(String value) {
 
   @Override
   public String toString() {
-    return "Headers{" + "isDefaultService=" + isDefaultService + ", keyValues=" + keyValues + '}';
+    return "Headers{"
+        + "isDefaultService="
+        + isDefaultService
+        + ", timeoutMillis="
+        + timeoutMillis
+        + ", retryPolicy="
+        + retryPolicy
+        + ", keyValues="
+        + keyValues
+        + '}';
   }
 
   public Headers.Builder toBuilder() {
@@ -162,7 +182,7 @@ public static Headers create(boolean isDefaultService, String... headers) {
     if (headers.length == 0) {
       return isDefaultService ? DEFAULT_SERVICE : EMPTY;
     }
-    return new Headers(isDefaultService, 0, Arrays.asList(headers), serializedSize);
+    return new Headers(isDefaultService, 0, null, Arrays.asList(headers), serializedSize);
   }
 
   public static Headers empty() {
@@ -178,7 +198,13 @@ public static Headers withTimeout(long timeoutMillis) {
     if (timeoutMillis == 0) {
       return EMPTY;
     }
-    return new Headers(false, timeoutMillis, Collections.emptyList(), 0);
+    return new Headers(false, timeoutMillis, null, Collections.emptyList(), 0);
+  }
+
+  public static Headers withRetry(long timeoutMillis, RetryPolicy retryPolicy) {
+    requireNonNegative(timeoutMillis, "timeoutMillis");
+    Objects.requireNonNull(retryPolicy, "retryPolicy");
+    return new Headers(false, timeoutMillis, retryPolicy, Collections.emptyList(), 0);
   }
 
   public static Headers.Builder newBuilder() {
@@ -194,7 +220,7 @@ static Headers create(List<String> headers) {
     if (headers.isEmpty()) {
       return EMPTY;
     }
-    return new Headers(false, 0, headers, serializedSize);
+    return new Headers(false, 0, null, headers, serializedSize);
   }
 
   ByteBuf cache() {
@@ -220,6 +246,7 @@ public static final class Builder {
     private boolean isDefaultService;
     private long timeoutMillis;
     private int serializedSize;
+    private RetryPolicy retryPolicy;
 
     private Builder(int size, List<String> headers) {
       int length = headers.size();
@@ -246,6 +273,11 @@ public Builder timeout(long timeoutMillis) {
       return this;
     }
 
+    public Builder retryPolicy(RetryPolicy retryPolicy) {
+      this.retryPolicy = Objects.requireNonNull(retryPolicy, "retryPolicy");
+      return this;
+    }
+
     public Builder add(String name, String value) {
       requireValidKeySize(name, " name");
       requireValidValueSize(value, " value");
@@ -299,16 +331,90 @@ public Builder remove(String name, String value) {
     }
 
     public Headers build() {
-      return new Headers(isDefaultService, timeoutMillis, nameValues, serializedSize);
+      return new Headers(isDefaultService, timeoutMillis, retryPolicy, nameValues, serializedSize);
     }
   }
 
-  private static String requireNonEmpty(String seq, String message) {
-    Objects.requireNonNull(seq, message);
-    if (seq.length() == 0) {
-      throw new IllegalArgumentException(message + " must be non-empty");
+  /**
+   * Configures automatic retry of failed requests. Intended mostly for RPC implementations lacking
+   * API support for retries, e.g. futures, grpc-stub.
+   */
+  public static final class RetryPolicy {
+    private final int maxAttempts;
+    private final float initialBackoff;
+    private final float maxBackoff;
+    private final int backoffMultiplier;
+
+    private RetryPolicy(
+        int maxAttempts, float initialBackoff, float maxBackoff, int backoffMultiplier) {
+      this.maxAttempts = maxAttempts;
+      this.initialBackoff = initialBackoff;
+      this.maxBackoff = maxBackoff;
+      this.backoffMultiplier = backoffMultiplier;
+    }
+
+    public static RetryPolicy create(
+        int maxAttempts, long initialBackoffMillis, long maxBackoffMillis, int backoffMultiplier) {
+      return new RetryPolicy(
+          requirePositive(maxAttempts, "maxAttempts"),
+          requireNonNegative(initialBackoffMillis, "initialBackoffMillis"),
+          requireNonNegative(maxBackoffMillis, "maxBackoffMillis"),
+          requireNonNegative(backoffMultiplier, "backoffMultiplier"));
+    }
+
+    public static RetryPolicy.Builder newBuilder() {
+      return new Builder();
+    }
+
+    @Override
+    public String toString() {
+      return "RetryPolicy{"
+          + "maxAttempts="
+          + maxAttempts
+          + ", initialBackoff="
+          + initialBackoff
+          + ", maxBackoff="
+          + maxBackoff
+          + ", backoffMultiplier="
+          + backoffMultiplier
+          + '}';
+    }
+
+    public static final class Builder {
+      private int maxAttempts;
+      private long initialBackoff;
+      private long maxBackoff;
+      private int backoffMultiplier;
+
+      public Builder() {}
+
+      public Builder maxAttempts(int maxAttempts) {
+        this.maxAttempts = requirePositive(maxAttempts, "maxAttempts");
+        return this;
+      }
+
+      public Builder initialBackoff(long initialBackoffMillis) {
+        this.initialBackoff = requireNonNegative(initialBackoffMillis, "initialBackoffMillis");
+        return this;
+      }
+
+      public Builder maxBackoff(long maxBackoffMillis) {
+        this.maxBackoff = requireNonNegative(maxBackoffMillis, "maxBackoffMillis");
+        return this;
+      }
+
+      public Builder backoffMultiplier(int backoffMultiplier) {
+        this.backoffMultiplier = requireNonNegative(backoffMultiplier, "backoffMultiplier");
+        return this;
+      }
+
+      public RetryPolicy build() {
+        if (initialBackoff > maxBackoff) {
+          throw new IllegalArgumentException("initialBackoff exceeds maxBackoff");
+        }
+        return new RetryPolicy(maxAttempts, initialBackoff, maxBackoff, backoffMultiplier);
+      }
     }
-    return seq;
   }
 
   private static int requireValid(List<String> keyValues, String message) {
@@ -344,6 +450,20 @@ private static long requireNonNegative(long value, String message) {
     return value;
   }
 
+  private static int requireNonNegative(int value, String message) {
+    if (value < 0) {
+      throw new IllegalArgumentException(message + " must be non-negative");
+    }
+    return value;
+  }
+
+  private static int requirePositive(int value, String message) {
+    if (value <= 0) {
+      throw new IllegalArgumentException(message + " must be positive");
+    }
+    return value;
+  }
+
   private static int requireValid(String[] keyValues, String message) {
     Objects.requireNonNull(keyValues, "keyValues");
     int length = keyValues.length;