Return Retry-After time to clients when they are rate limited (#421)

* Return Retry-After time to clients when they are rate limited

* Update based on feedback

- New exception type that is mapped differently
- Always report time until allowed on rate limits
- Consume and transform into a differnt exception if we think it will be
  allowed later

service/src/main/java/org/whispersystems/textsecuregcm/limits/LeakyBucket.java

@@ -48,6 +48,18 @@ public class LeakyBucket {
                     (int)Math.floor(this.spaceRemaining + (elapsedTime * this.leakRatePerMillis)));
   }
 
+  public long getMillisUntilSpace(double amount) {
+    int currentSpaceRemaining = getUpdatedSpaceRemaining();
+    if (currentSpaceRemaining >= amount) {
+      return 0;
+    } else if (amount > this.bucketSize) {
+      // This shouldn't happen today but if so we should bubble this to the clients somehow
+      return -1;
+    } else {
+      return (long)Math.ceil(amount - currentSpaceRemaining / this.leakRatePerMillis);
+    }
+  }
+
   public String serialize(ObjectMapper mapper) throws JsonProcessingException {
     return mapper.writeValueAsString(new LeakyBucketEntity(bucketSize, leakRatePerMillis, spaceRemaining, lastUpdateTimeMillis));
   }

service/src/main/java/org/whispersystems/textsecuregcm/limits/RateLimiter.java

@@ -28,7 +28,7 @@ public class RateLimiter {
   private final ObjectMapper mapper = SystemMapper.getMapper();
 
   private   final Meter                     meter;
-  private   final Timer                     validateTimer;
+  protected final Timer                     validateTimer;
   protected final FaultTolerantRedisCluster cacheCluster;
   protected final String                    name;
   private   final int                       bucketSize;
@@ -66,7 +66,7 @@ public class RateLimiter {
         setBucket(key, bucket);
       } else {
         meter.mark();
-        throw new RateLimitExceededException(key + " , " + amount);
+        throw new RateLimitExceededException(key + " , " + amount, bucket.getMillisUntilSpace(amount));
       }
     }
   }
@@ -87,7 +87,7 @@ public class RateLimiter {
     return leakRatePerMinute;
   }
 
-  private void setBucket(String key, LeakyBucket bucket) {
+  protected void setBucket(String key, LeakyBucket bucket) {
     try {
       final String serialized = bucket.serialize(mapper);
 
@@ -97,7 +97,7 @@ public class RateLimiter {
     }
   }
 
-  private LeakyBucket getBucket(String key) {
+  protected LeakyBucket getBucket(String key) {
     try {
       final String serialized = cacheCluster.withCluster(connection -> connection.sync().get(getBucketName(key)));