Skip to content

Commit 39e3261

Browse files
kisagakisaga
committed
added disconnect functionality when stream is empty for 20 secs and some exception handling
1 parent 9d28fe1 commit 39e3261

7 files changed

Lines changed: 61 additions & 34 deletions

File tree

‎examples/src/main/java/com/twitter/clientlib/HelloWorldStreaming.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ public static void main(String[] args) {
6565

6666
class Responder implements TweetsStreamListener {
6767
@Override
68-
public synchronized void actionOnTweetsStream(StreamingTweet streamingTweet) {
68+
public synchronized void onTweetArrival(StreamingTweet streamingTweet) {
6969
if(streamingTweet == null) {
7070
System.err.println("Error: actionOnTweetsStream - streamingTweet is null ");
7171
return;

‎src/main/java/com/twitter/clientlib/api/TweetsApi.java‎

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -23,25 +23,16 @@
2323
package com.twitter.clientlib.api;
2424

2525
import com.twitter.clientlib.ApiCallback;
26-
import com.twitter.clientlib.ApiClient;
27-
import com.twitter.clientlib.auth.*;
2826
import com.twitter.clientlib.ApiException;
2927
import com.twitter.clientlib.ApiResponse;
30-
import com.twitter.clientlib.Configuration;
3128
import com.twitter.clientlib.Pair;
32-
import com.twitter.clientlib.ProgressRequestBody;
33-
import com.twitter.clientlib.ProgressResponseBody;
3429

35-
import com.github.scribejava.core.model.OAuth2AccessToken;
3630
import com.google.gson.reflect.TypeToken;
3731

38-
import java.io.IOException;
39-
4032

4133
import com.twitter.clientlib.model.AddOrDeleteRulesRequest;
4234
import com.twitter.clientlib.model.AddOrDeleteRulesResponse;
4335
import com.twitter.clientlib.model.CreateTweetRequest;
44-
import com.twitter.clientlib.model.Error;
4536
import com.twitter.clientlib.model.FilteredStreamingTweet;
4637
import com.twitter.clientlib.model.GenericTweetsTimelineResponse;
4738
import com.twitter.clientlib.model.GetRulesResponse;
@@ -52,7 +43,7 @@
5243
import com.twitter.clientlib.model.MultiTweetLookupResponse;
5344
import com.twitter.clientlib.model.MultiUserLookupResponse;
5445
import java.time.OffsetDateTime;
55-
import com.twitter.clientlib.model.Problem;
46+
5647
import com.twitter.clientlib.model.QuoteTweetLookupResponse;
5748
import com.twitter.clientlib.model.SingleTweetLookupResponse;
5849
import com.twitter.clientlib.model.StreamingTweet;
@@ -75,11 +66,8 @@
7566
import java.util.List;
7667
import java.util.Map;
7768
import java.util.Set;
78-
import java.io.InputStream;
79-
import javax.ws.rs.core.GenericType;
8069

8170
import okio.BufferedSource;
82-
import org.apache.commons.lang3.StringUtils;
8371

8472
public class TweetsApi extends ApiCommon {
8573

‎src/main/java/com/twitter/clientlib/exceptions/AuthenticationException.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
package com.twitter.clientlib.exceptions;
2+
3+
public class AuthenticationException extends RuntimeException {
4+
5+
public AuthenticationException(String message) {
6+
super(message);
7+
}
8+
9+
public AuthenticationException(String message, Throwable cause) {
10+
super(message, cause);
11+
}
12+
}

‎src/main/java/com/twitter/clientlib/exceptions/EmptyStreamTimeoutException.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package com.twitter.clientlib.exceptions;
2+
3+
public class EmptyStreamTimeoutException extends RuntimeException {
4+
public EmptyStreamTimeoutException(String message) {
5+
super(message);
6+
}
7+
}

‎src/main/java/com/twitter/clientlib/stream/TweetsStreamExecutor.java‎

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
package com.twitter.clientlib.stream;
2424

2525

26+
import java.io.IOException;
2627
import java.util.ArrayList;
2728
import java.util.List;
2829
import java.util.concurrent.BlockingQueue;
@@ -36,10 +37,14 @@
3637
import com.fasterxml.jackson.databind.JsonMappingException;
3738
import com.fasterxml.jackson.databind.ObjectMapper;
3839

40+
import com.twitter.clientlib.exceptions.EmptyStreamTimeoutException;
3941
import com.twitter.clientlib.model.StreamingTweet;
4042
import okio.BufferedSource;
4143

4244
public class TweetsStreamExecutor {
45+
46+
private static final long EMPTY_STREAM_TIMEOUT = 20000;
47+
private static final int POLL_WAIT = 5;
4348
private volatile BlockingQueue<String> rawTweets;
4449
private volatile BlockingQueue<StreamingTweet> tweets;
4550
private volatile boolean isRunning = true;
@@ -90,9 +95,12 @@ public synchronized void shutdown() {
9095
shutDownServices();
9196
try {
9297
terminateServices();
98+
stream.close();
9399
} catch (InterruptedException ie) {
94100
shutDownServices();
95101
Thread.currentThread().interrupt();
102+
} catch (IOException e) {
103+
96104
}
97105
System.out.println("TweetsStreamListenersExecutor is shutting down.");
98106
}
@@ -109,9 +117,9 @@ private void terminateServices() throws InterruptedException {
109117
terminateService(listenersService);
110118
}
111119
private void terminateService(ExecutorService executorService) throws InterruptedException {
112-
if (!executorService.awaitTermination(3, TimeUnit.SECONDS)) {
120+
if (!executorService.awaitTermination(1500, TimeUnit.MILLISECONDS)) {
113121
executorService.shutdownNow();
114-
if (!executorService.awaitTermination(3, TimeUnit.SECONDS))
122+
if (!executorService.awaitTermination(1500, TimeUnit.MILLISECONDS))
115123
System.err.println("Pool did not terminate");
116124
}
117125
}
@@ -126,19 +134,32 @@ public void run() {
126134
public void queueTweets() {
127135
String line = null;
128136
try {
137+
boolean emptyResponse = false;
138+
long firstEmptyResponseMillis = 0;
139+
long lastEmptyReponseMillis;
129140
while (isRunning) {
130141
line = stream.readUtf8Line();
131142
if(line == null || line.isEmpty()) {
143+
if(!emptyResponse) {
144+
firstEmptyResponseMillis = System.currentTimeMillis();
145+
emptyResponse = true;
146+
} else {
147+
lastEmptyReponseMillis = System.currentTimeMillis();
148+
if(lastEmptyReponseMillis - firstEmptyResponseMillis > EMPTY_STREAM_TIMEOUT) {
149+
throw new EmptyStreamTimeoutException(String.format("Stream was empty for %d seconds consecutively", EMPTY_STREAM_TIMEOUT));
150+
}
151+
}
132152
continue;
133153
}
154+
emptyResponse = false;
134155
try {
135156
rawTweets.put(line);
136-
} catch (Exception interExcep) {
137-
interExcep.printStackTrace();
157+
} catch (Exception ignore) {
158+
138159
}
139160
}
140161
} catch (Exception e) {
141-
e.printStackTrace();
162+
System.out.println("Something went wrong. Closing stream... " + e.getMessage());
142163
shutdown();
143164
}
144165
}
@@ -156,15 +177,14 @@ private DeserializeTweetsTask() {
156177
public void run() {
157178
while (isRunning) {
158179
try {
159-
String rawTweet = rawTweets.take();
180+
String rawTweet = rawTweets.poll(POLL_WAIT, TimeUnit.MILLISECONDS);
181+
if (rawTweet == null) continue;
160182
StreamingTweet tweet = objectMapper.readValue(rawTweet, StreamingTweet.class);
161183
tweets.put(tweet);
162184
} catch (InterruptedException e) {
163-
System.out.println("Fail 1");
164-
} catch (JsonMappingException e) {
165-
System.out.println("Fail 2");
185+
166186
} catch (JsonProcessingException e) {
167-
System.out.println("Fail 3");
187+
System.out.println("debug log here");
168188
}
169189
}
170190
}
@@ -181,9 +201,10 @@ private void processTweets() {
181201

182202
while (isRunning) {
183203
try {
184-
streamingTweet = tweets.take();
204+
streamingTweet = tweets.poll(POLL_WAIT, TimeUnit.MILLISECONDS);
205+
if(streamingTweet == null) continue;
185206
for (TweetsStreamListener listener : listeners) {
186-
listener.actionOnTweetsStream(streamingTweet);
207+
listener.onTweetArrival(streamingTweet);
187208
}
188209
tweetsCount++;
189210
if(tweetsCount == tweetsLimit) {
@@ -194,7 +215,7 @@ private void processTweets() {
194215
shutdown();
195216
}
196217
} catch (InterruptedException e) {
197-
System.out.println("processTweets: Fail 1");
218+
198219
}
199220

200221
}

‎src/main/java/com/twitter/clientlib/stream/TweetsStreamListener.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,5 +25,5 @@
2525
import com.twitter.clientlib.model.StreamingTweet;
2626

2727
public interface TweetsStreamListener {
28-
void actionOnTweetsStream(StreamingTweet streamingTweet);
28+
void onTweetArrival(StreamingTweet streamingTweet);
2929
}

‎src/main/java/com/twitter/clientlib/stream/TwitterStream.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,10 @@
44
import com.twitter.clientlib.ApiException;
55
import com.twitter.clientlib.TwitterCredentialsBearer;
66
import com.twitter.clientlib.api.TweetsApi;
7+
import com.twitter.clientlib.exceptions.AuthenticationException;
78
import com.twitter.clientlib.query.StreamQueryParameters;
89
import okio.BufferedSource;
910

10-
import java.io.InputStream;
1111
import java.util.LinkedList;
1212
import java.util.List;
1313

@@ -38,15 +38,14 @@ public void sampleStream(StreamQueryParameters streamParameters) {
3838
listeners.forEach(executor::addListener);
3939
executor.start();
4040
} catch (ApiException e) {
41-
System.err.println("Status code: " + e.getCode());
42-
System.err.println("Reason: " + e.getResponseBody());
43-
System.err.println("Response headers: " + e.getResponseHeaders());
44-
e.printStackTrace();
41+
if(e.getCode() == 401) {
42+
throw new AuthenticationException("Not authenticated. Please check the credentials", e);
43+
}
4544
}
4645
}
4746

4847
private void initBasePath() {
49-
String basePath = "http://localhost:8080";
48+
String basePath = System.getenv("TWITTER_API_BASE_PATH");
5049
apiClient.setBasePath(basePath != null ? basePath : "https://api.twitter.com");
5150
}
5251

0 commit comments

Comments
 (0)