Skip to content

Commit 825d8fc

Browse files
simplestephsimplesteph
committed
separated the GitHub API Http Client for better readability
1 parent 6015618 commit 825d8fc

3 files changed

Lines changed: 120 additions & 97 deletions

File tree

‎src/main/java/com/simplesteph/kafka/GitHubAPIHttpClient.java‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
package com.simplesteph.kafka;
2+
3+
import com.mashape.unirest.http.Headers;
4+
import com.mashape.unirest.http.HttpResponse;
5+
import com.mashape.unirest.http.JsonNode;
6+
import com.mashape.unirest.http.Unirest;
7+
import com.mashape.unirest.http.exceptions.UnirestException;
8+
import com.mashape.unirest.request.GetRequest;
9+
import org.apache.kafka.connect.errors.ConnectException;
10+
import org.json.JSONArray;
11+
import org.slf4j.Logger;
12+
import org.slf4j.LoggerFactory;
13+
14+
import java.time.Instant;
15+
import java.time.LocalDateTime;
16+
import java.time.ZoneOffset;
17+
18+
// GitHubHttpAPIClient used to launch HTTP Get requests
19+
public class GitHubAPIHttpClient {
20+
21+
private static final Logger log = LoggerFactory.getLogger(GitHubAPIHttpClient.class);
22+
23+
// for efficient http requests
24+
private Integer XRateLimit = 9999;
25+
private Integer XRateRemaining = 9999;
26+
private long XRateReset = Instant.MAX.getEpochSecond();
27+
28+
GitHubSourceConnectorConfig config;
29+
30+
public GitHubAPIHttpClient(GitHubSourceConnectorConfig config){
31+
this.config = config;
32+
}
33+
34+
protected JSONArray getNextIssues(Integer page, Instant since) throws InterruptedException {
35+
36+
HttpResponse<JsonNode> jsonResponse;
37+
try {
38+
jsonResponse = getNextIssuesAPI(page, since);
39+
40+
// deal with headers in any case
41+
Headers headers = jsonResponse.getHeaders();
42+
XRateLimit = Integer.valueOf(headers.getFirst("X-RateLimit-Limit"));
43+
XRateRemaining = Integer.valueOf(headers.getFirst("X-RateLimit-Remaining"));
44+
XRateReset = Integer.valueOf(headers.getFirst("X-RateLimit-Reset"));
45+
switch (jsonResponse.getStatus()){
46+
case 200:
47+
return jsonResponse.getBody().getArray();
48+
case 401:
49+
throw new ConnectException("Bad GitHub credentials provided, please edit your config");
50+
case 403:
51+
// we have issues too many requests.
52+
log.info(jsonResponse.getBody().getObject().getString("message"));
53+
log.info(String.format("Your rate limit is %s", XRateLimit));
54+
log.info(String.format("Your remaining calls is %s", XRateRemaining));
55+
log.info(String.format("The limit will reset at %s",
56+
LocalDateTime.ofInstant(Instant.ofEpochSecond(XRateReset), ZoneOffset.systemDefault())));
57+
long sleepTime = XRateReset - Instant.now().getEpochSecond();
58+
log.info(String.format("Sleeping for %s seconds", sleepTime ));
59+
Thread.sleep(1000 * sleepTime);
60+
return getNextIssues(page, since);
61+
default:
62+
log.error(String.valueOf(jsonResponse.getStatus()));
63+
log.error(jsonResponse.getBody().toString());
64+
log.error(jsonResponse.getHeaders().toString());
65+
log.error("Unknown error: Sleeping 5 seconds " +
66+
"before re-trying");
67+
Thread.sleep(5000L);
68+
return getNextIssues(page, since);
69+
}
70+
} catch (UnirestException e) {
71+
e.printStackTrace();
72+
Thread.sleep(5000L);
73+
return new JSONArray();
74+
}
75+
}
76+
77+
protected HttpResponse<JsonNode> getNextIssuesAPI(Integer page, Instant since) throws UnirestException {
78+
GetRequest unirest = Unirest.get(constructUrl(page, since));
79+
if (!config.getAuthUsername().isEmpty() && !config.getAuthPassword().isEmpty() ){
80+
unirest = unirest.basicAuth(config.getAuthUsername(), config.getAuthPassword());
81+
}
82+
log.debug(String.format("GET %s", unirest.getUrl()));
83+
return unirest.asJson();
84+
}
85+
86+
protected String constructUrl(Integer page, Instant since){
87+
return String.format(
88+
"https://api.github.com/repos/%s/%s/issues?page=%s&per_page=%s&since=%s&state=all&direction=asc&sort=updated",
89+
config.getOwnerConfig(),
90+
config.getRepoConfig(),
91+
page,
92+
config.getBatchSize(),
93+
since.toString());
94+
}
95+
96+
public void sleep() throws InterruptedException {
97+
long sleepTime = (long) Math.ceil(
98+
(double) (XRateReset - Instant.now().getEpochSecond()) / XRateRemaining);
99+
log.debug(String.format("Sleeping for %s seconds", sleepTime ));
100+
Thread.sleep(1000 * sleepTime);
101+
}
102+
103+
public void sleepIfNeed() throws InterruptedException {
104+
// Sleep if needed
105+
if (XRateRemaining <= 10 && XRateRemaining > 0) {
106+
log.info(String.format("Approaching limit soon, you have %s requests left", XRateRemaining));
107+
sleep();
108+
}
109+
}
110+
}

‎src/main/java/com/simplesteph/kafka/GitHubSourceTask.java‎

Lines changed: 7 additions & 95 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,10 @@
11
package com.simplesteph.kafka;
22

3-
import com.mashape.unirest.http.Headers;
4-
import com.mashape.unirest.http.HttpResponse;
5-
import com.mashape.unirest.http.JsonNode;
6-
import com.mashape.unirest.http.Unirest;
7-
import com.mashape.unirest.http.exceptions.UnirestException;
8-
import com.mashape.unirest.request.GetRequest;
93
import com.simplesteph.kafka.model.Issue;
104
import com.simplesteph.kafka.model.PullRequest;
115
import com.simplesteph.kafka.model.User;
126
import com.simplesteph.kafka.utils.DateUtils;
137
import org.apache.kafka.connect.data.Struct;
14-
import org.apache.kafka.connect.errors.ConnectException;
158
import org.apache.kafka.connect.source.SourceRecord;
169
import org.apache.kafka.connect.source.SourceTask;
1710
import org.json.JSONArray;
@@ -20,27 +13,20 @@
2013
import org.slf4j.LoggerFactory;
2114

2215
import java.time.Instant;
23-
import java.time.LocalDateTime;
24-
import java.time.ZoneOffset;
2516
import java.util.*;
2617
import static com.simplesteph.kafka.GitHubSchemas.*;
2718

2819

2920
public class GitHubSourceTask extends SourceTask {
30-
static final Logger log = LoggerFactory.getLogger(GitHubSourceTask.class);
21+
private static final Logger log = LoggerFactory.getLogger(GitHubSourceTask.class);
3122
public GitHubSourceConnectorConfig config;
3223

3324
protected Instant nextQuerySince;
3425
protected Integer lastIssueNumber;
3526
protected Integer nextPageToVisit = 1;
3627
protected Instant lastUpdatedAt;
3728

38-
GitHubHttpAPIClient gitHubHttpAPIClient = new GitHubHttpAPIClient();
39-
40-
// for efficient http requests
41-
private Integer XRateLimit = 9999;
42-
private Integer XRateRemaining = 9999;
43-
private long XRateReset = Instant.MAX.getEpochSecond();
29+
GitHubAPIHttpClient gitHubHttpAPIClient;
4430

4531
@Override
4632
public String version() {
@@ -52,6 +38,7 @@ public void start(Map<String, String> map) {
5238
//Do things here that are required to start your task. This could be open a connection to a database, etc.
5339
config = new GitHubSourceConnectorConfig(map);
5440
initializeLastVariables();
41+
gitHubHttpAPIClient = new GitHubAPIHttpClient(config);
5542
}
5643

5744
private void initializeLastVariables(){
@@ -77,24 +64,15 @@ private void initializeLastVariables(){
7764
}
7865
}
7966

80-
public void sleep() throws InterruptedException {
81-
long sleepTime = (long) Math.ceil(
82-
(double) (XRateReset - Instant.now().getEpochSecond()) / XRateRemaining);
83-
log.debug(String.format("Sleeping for %s seconds", sleepTime ));
84-
Thread.sleep(1000 * sleepTime);
85-
}
67+
8668

8769
@Override
8870
public List<SourceRecord> poll() throws InterruptedException {
89-
// Sleep if needed
90-
if (XRateRemaining <= 10 && XRateRemaining > 0){
91-
log.info(String.format("Approaching limit soon, you have %s requests left", XRateRemaining));
92-
sleep();
93-
}
71+
gitHubHttpAPIClient.sleepIfNeed();
9472

9573
// fetch data
9674
final ArrayList<SourceRecord> records = new ArrayList<>();
97-
JSONArray issues = gitHubHttpAPIClient.getNextIssues();
75+
JSONArray issues = gitHubHttpAPIClient.getNextIssues(nextPageToVisit, nextQuerySince);
9876
// we'll count how many results we get with i
9977
int i = 0;
10078
for (Object obj : issues) {
@@ -112,7 +90,7 @@ public List<SourceRecord> poll() throws InterruptedException {
11290
else {
11391
nextQuerySince = lastUpdatedAt.plusSeconds(1);
11492
nextPageToVisit = 1;
115-
sleep();
93+
gitHubHttpAPIClient.sleep();
11694
}
11795
return records;
11896
}
@@ -190,70 +168,4 @@ private Struct buildRecordValue(Issue issue){
190168
return valueStruct;
191169
}
192170

193-
// GitHubHttpAPIClient used to launch HTTP Get requests
194-
public class GitHubHttpAPIClient {
195-
196-
protected JSONArray getNextIssues() throws InterruptedException {
197-
198-
HttpResponse<JsonNode> jsonResponse;
199-
try {
200-
jsonResponse = getNextIssuesAPI();
201-
202-
// deal with headers in any case
203-
Headers headers = jsonResponse.getHeaders();
204-
XRateLimit = Integer.valueOf(headers.getFirst("X-RateLimit-Limit"));
205-
XRateRemaining = Integer.valueOf(headers.getFirst("X-RateLimit-Remaining"));
206-
XRateReset = Integer.valueOf(headers.getFirst("X-RateLimit-Reset"));
207-
switch (jsonResponse.getStatus()){
208-
case 200:
209-
return jsonResponse.getBody().getArray();
210-
case 401:
211-
throw new ConnectException("Bad GitHub credentials provided, please edit your config");
212-
case 403:
213-
// we have issues too many requests.
214-
log.info(jsonResponse.getBody().getObject().getString("message"));
215-
log.info(String.format("Your rate limit is %s", XRateLimit));
216-
log.info(String.format("Your remaining calls is %s", XRateRemaining));
217-
log.info(String.format("The limit will reset at %s",
218-
LocalDateTime.ofInstant(Instant.ofEpochSecond(XRateReset), ZoneOffset.systemDefault())));
219-
long sleepTime = XRateReset - Instant.now().getEpochSecond();
220-
log.info(String.format("Sleeping for %s seconds", sleepTime ));
221-
Thread.sleep(1000 * sleepTime);
222-
return getNextIssues();
223-
default:
224-
log.error(String.valueOf(jsonResponse.getStatus()));
225-
log.error(jsonResponse.getBody().toString());
226-
log.error(jsonResponse.getHeaders().toString());
227-
log.error("Unknown error: Sleeping 5 seconds " +
228-
"before re-trying");
229-
Thread.sleep(5000L);
230-
return getNextIssues();
231-
}
232-
} catch (UnirestException e) {
233-
e.printStackTrace();
234-
Thread.sleep(5000L);
235-
return new JSONArray();
236-
}
237-
}
238-
239-
protected HttpResponse<JsonNode> getNextIssuesAPI() throws UnirestException {
240-
GetRequest unirest = Unirest.get(constructUrl());
241-
if (!config.getAuthUsername().isEmpty() && !config.getAuthPassword().isEmpty() ){
242-
unirest = unirest.basicAuth(config.getAuthUsername(), config.getAuthPassword());
243-
}
244-
log.debug(String.format("GET %s", unirest.getUrl()));
245-
return unirest.asJson();
246-
}
247-
248-
protected String constructUrl(){
249-
return String.format(
250-
"https://api.github.com/repos/%s/%s/issues?page=%s&per_page=%s&since=%s&state=all&direction=asc&sort=updated",
251-
config.getOwnerConfig(),
252-
config.getRepoConfig(),
253-
nextPageToVisit,
254-
config.getBatchSize(),
255-
nextQuerySince.toString());
256-
}
257-
}
258-
259171
}

‎src/test/java/com/simplesteph/kafka/GitHubSourceTaskTest.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,9 +35,10 @@ public void test() throws UnirestException {
3535
gitHubSourceTask.config = new GitHubSourceConnectorConfig(initialConfig());
3636
gitHubSourceTask.nextPageToVisit = 1;
3737
gitHubSourceTask.nextQuerySince = Instant.parse("2017-01-01T00:00:00Z");
38-
String url = gitHubSourceTask.gitHubHttpAPIClient.constructUrl();
38+
gitHubSourceTask.gitHubHttpAPIClient = new GitHubAPIHttpClient(gitHubSourceTask.config);
39+
String url = gitHubSourceTask.gitHubHttpAPIClient.constructUrl(gitHubSourceTask.nextPageToVisit, gitHubSourceTask.nextQuerySince);
3940
System.out.println(url);
40-
HttpResponse<JsonNode> httpResponse = gitHubSourceTask.gitHubHttpAPIClient.getNextIssuesAPI();
41+
HttpResponse<JsonNode> httpResponse = gitHubSourceTask.gitHubHttpAPIClient.getNextIssuesAPI(gitHubSourceTask.nextPageToVisit, gitHubSourceTask.nextQuerySince);
4142
if (httpResponse.getStatus() != 403) {
4243
assert (httpResponse.getStatus() == 200);
4344
Set<String> headers = httpResponse.getHeaders().keySet();

0 commit comments

Comments
 (0)