1. import java.util.HashMap;
  2. import java.util.Map;
  3. import java.util.concurrent.ConcurrentHashMap;
  4. import java.util.concurrent.atomic.AtomicInteger;
  6. public class ApiPayloadWrapper {
  8. private static final ConcurrentHashMap<String, RateLimiter> rateLimiters = new ConcurrentHashMap<>();
  10. public static <T> T wrapPayload(String apiEndpoint, Runnable payloadHandler, int maxRequestsPerSecond) {
  11. // Get or create a rate limiter for the endpoint.
  12. RateLimiter limiter = rateLimiters.computeIfAbsent(apiEndpoint, k -> new RateLimiter(maxRequestsPerSecond));
  14. // Acquire a token from the rate limiter. Blocks until a token is available.
  15. try {
  16. limiter.acquire();
  17. } catch (InterruptedException e) {
  18. Thread.currentThread().interrupt(); // Restore interrupted status
  19. throw new RuntimeException("Rate limit interrupted", e);
  20. }
  22. // Execute the payload handler.
  23. try {
  24. payloadHandler.run();
  25. } catch (Exception e) {
  26. //Wrap the exception with a maintenance error.
  27. throw new MaintenanceError("API Payload Error: " + e.getMessage(), e);
  28. }
  30. return null; // Return null or the result of the payload handler, if applicable.
  31. }
  33. public static class RateLimiter {
  34. private final int maxRequestsPerSecond;
  35. private final AtomicInteger requestCount = new AtomicInteger(0);
  36. private long lastResetTime = System.currentTimeMillis();
  38. public RateLimiter(int maxRequestsPerSecond) {
  39. this.maxRequestsPerSecond = maxRequestsPerSecond;
  40. }
  42. public synchronized boolean acquire() throws InterruptedException {
  43. long currentTime = System.currentTimeMillis();
  44. if (currentTime - lastResetTime > 1000) {
  45. // Reset the counter if a second has passed.
  46. requestCount.set(0);
  47. lastResetTime = currentTime;
  48. }
  50. if (requestCount.get() < maxRequestsPerSecond) {
  51. requestCount.incrementAndGet();
  52. return true;
  53. } else {
  54. // Wait until a token is available.
  55. long sleepTime = 1000 - (currentTime - lastResetTime) % 1000; //Ensure we don't wait longer than 1s
  56. Thread.sleep(sleepTime);
  57. return acquire(); //Recursive call to try again
  58. }
  59. }
  60. }
  62. public static class MaintenanceError extends RuntimeException {
  63. private final Throwable cause;
  65. public MaintenanceError(String message, Throwable cause) {
  66. super(message);
  67. this.cause = cause;
  68. }
  70. public Throwable getCause() {
  71. return cause;
  72. }
  73. }
  75. public static void main(String[] args) {
  76. // Example Usage
  77. try {
  78. // Simulate an API endpoint.
  79. Runnable apiCall = () -> {
  80. System.out.println("Making API call...");
  81. //Simulate a possible error
  82. if(Math.random() < 0.2){
  83. throw new RuntimeException("Simulated API error");
  84. }
  85. System.out.println("API call completed.");
  86. };
  88. // Wrap the API call with a rate limiter.
  89. T result = wrapPayload("exampleEndpoint", apiCall, 5); // 5 requests per second
  91. if (result != null) {
  92. System.out.println("API call result: " + result);
  93. }
  94. } catch (MaintenanceError e) {
  95. System.err.println("Maintenance Error: " + e.getMessage());
  96. System.err.println("Caused by: " + e.getCause().getMessage());
  97. }
  98. }
  99. }

Add your comment