Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: automatically handle helix rate limits (#306)
- Loading branch information
Showing
6 changed files
with
211 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
48 changes: 48 additions & 0 deletions
48
rest-helix/src/main/java/com/github/twitch4j/helix/interceptor/TwitchHelixDecoder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| package com.github.twitch4j.helix.interceptor; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import feign.Response; | ||
| import feign.jackson.JacksonDecoder; | ||
|
|
||
| import java.io.IOException; | ||
| import java.lang.reflect.Type; | ||
| import java.util.Collection; | ||
|
|
||
| import static com.github.twitch4j.helix.interceptor.TwitchHelixClientIdInterceptor.AUTH_HEADER; | ||
| import static com.github.twitch4j.helix.interceptor.TwitchHelixClientIdInterceptor.BEARER_PREFIX; | ||
|
|
||
| public class TwitchHelixDecoder extends JacksonDecoder { | ||
|
|
||
| public static final String REMAINING_HEADER = "Ratelimit-Remaining"; | ||
|
|
||
| private final TwitchHelixClientIdInterceptor interceptor; | ||
|
|
||
| public TwitchHelixDecoder(ObjectMapper mapper, TwitchHelixClientIdInterceptor interceptor) { | ||
| super(mapper); | ||
| this.interceptor = interceptor; | ||
| } | ||
|
|
||
| @Override | ||
| public Object decode(Response response, Type type) throws IOException { | ||
| // track rate limit for token | ||
| String token = singleFirst(response.request().headers().get(AUTH_HEADER)); | ||
| if (token != null && token.startsWith(BEARER_PREFIX)) { | ||
| String remaining = singleFirst(response.headers().get(REMAINING_HEADER)); | ||
| if (remaining != null) { | ||
| try { | ||
| interceptor.updateRemaining(token.substring(BEARER_PREFIX.length()), Integer.parseInt(remaining)); | ||
| } catch (Exception ignored) { | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // delegate to JacksonDecoder | ||
| return super.decode(response, type); | ||
| } | ||
|
|
||
| static String singleFirst(Collection<String> collection) { | ||
| if (collection == null || collection.size() != 1) return null; | ||
| return collection.toArray(new String[1])[0]; | ||
| } | ||
|
|
||
| } |
73 changes: 73 additions & 0 deletions
73
rest-helix/src/main/java/com/github/twitch4j/helix/interceptor/TwitchHelixHttpClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package com.github.twitch4j.helix.interceptor; | ||
|
|
||
| import com.github.philippheuer.credentialmanager.domain.OAuth2Credential; | ||
| import feign.Client; | ||
| import feign.Request; | ||
| import feign.Response; | ||
| import feign.okhttp.OkHttpClient; | ||
| import io.github.bucket4j.Bucket; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
| import java.util.concurrent.ScheduledThreadPoolExecutor; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.TimeoutException; | ||
|
|
||
| import static com.github.twitch4j.helix.interceptor.TwitchHelixClientIdInterceptor.AUTH_HEADER; | ||
| import static com.github.twitch4j.helix.interceptor.TwitchHelixClientIdInterceptor.BEARER_PREFIX; | ||
| import static com.github.twitch4j.helix.interceptor.TwitchHelixDecoder.singleFirst; | ||
|
|
||
| @Slf4j | ||
| public class TwitchHelixHttpClient implements Client { | ||
|
|
||
| private final Client client; | ||
| private final ScheduledExecutorService executor; | ||
| private final TwitchHelixClientIdInterceptor interceptor; | ||
| private final long timeout; | ||
|
|
||
| public TwitchHelixHttpClient(OkHttpClient client, ScheduledThreadPoolExecutor executor, TwitchHelixClientIdInterceptor interceptor, Integer timeout) { | ||
| this.client = client; | ||
| this.executor = executor; | ||
| this.interceptor = interceptor; | ||
| this.timeout = timeout == null ? 60 * 1000 : timeout.longValue(); | ||
| } | ||
|
|
||
| @Override | ||
| public Response execute(Request request, Request.Options options) throws IOException { | ||
| // Check whether this request should be delayed to conform to rate limits | ||
| String token = singleFirst(request.headers().get(AUTH_HEADER)); | ||
| if (token != null && token.startsWith(BEARER_PREFIX)) { | ||
| OAuth2Credential credential = interceptor.getAccessTokenCache().getIfPresent(token.substring(BEARER_PREFIX.length())); | ||
| if (credential != null) { | ||
| Bucket bucket = interceptor.getOrInitializeBucket(interceptor.getKey(credential)); | ||
| if (bucket.tryConsume(1)) { | ||
| // no delay needed | ||
| return client.execute(request, options); | ||
| } else { | ||
| try { | ||
| // effectively blocking, unfortunately | ||
| return bucket.asAsyncScheduler().consume(1, executor) | ||
| .thenApplyAsync(v -> { | ||
| try { | ||
| return client.execute(request, options); | ||
| } catch (IOException e) { | ||
| log.error("Helix API call execution failed", e); | ||
| return null; | ||
| } | ||
| }) | ||
| .get(timeout, TimeUnit.MILLISECONDS); | ||
| } catch (InterruptedException | ExecutionException | TimeoutException e) { | ||
| log.error("Throttled Helix API call timed-out before completion", e); | ||
| return null; | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Fallback: just run the http request | ||
| return client.execute(request, options); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters