Skip to content

Commit 5eb7ddd

Browse files
kisagakisaga
committed
created TwitterStream and moved executor from examples
Also, blocking queues were added and started parsing the raw responses to objects in parallel
1 parent 0386e59 commit 5eb7ddd

File tree

4 files changed

+123
-102
lines changed

4 files changed

+123
-102
lines changed

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

Lines changed: 9 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -23,17 +23,11 @@
2323
package com.twitter.clientlib;
2424

2525

26-
import java.util.HashSet;
27-
import java.util.Set;
28-
import java.io.InputStream;
29-
30-
import com.twitter.clientlib.ApiException;
31-
import com.twitter.clientlib.TwitterCredentialsBearer;
32-
import com.twitter.clientlib.TweetsStreamListenersExecutor;
33-
import com.twitter.clientlib.api.TwitterApi;
3426
import com.twitter.clientlib.model.*;
3527
import com.twitter.clientlib.query.StreamQueryParameters;
3628
import com.twitter.clientlib.query.model.TweetField;
29+
import com.twitter.clientlib.stream.TweetsStreamListener;
30+
import com.twitter.clientlib.stream.TwitterStream;
3731

3832

3933
public class HelloWorldStreaming {
@@ -46,58 +40,17 @@ public static void main(String[] args) {
4640
* to use the right credential object.
4741
*/
4842
TwitterCredentialsBearer credentials = new TwitterCredentialsBearer(System.getenv("TWITTER_BEARER_TOKEN"));
49-
TwitterApi apiInstance = new TwitterApi();
50-
apiInstance.setTwitterCredentials(credentials);
51-
52-
try {
53-
InputStream streamResult = apiInstance.tweets().sampleStream(
54-
new StreamQueryParameters.Builder()
55-
.withTweetFields(TweetField.AUTHOR_ID, TweetField.ID, TweetField.CREATED_AT)
56-
.build());
57-
// sampleStream with TweetsStreamListenersExecutor
58-
Responder responder = new Responder();
59-
TweetsStreamListenersExecutor tsle = new TweetsStreamListenersExecutor(streamResult);
60-
tsle.addListener(responder);
61-
tsle.executeListeners();
43+
TwitterStream twitterStream = new TwitterStream();
44+
twitterStream.setTwitterCredentials(credentials);
45+
twitterStream.addListener(new Responder());
46+
twitterStream.sampleStream(new StreamQueryParameters.Builder()
47+
.withTweetFields(TweetField.AUTHOR_ID, TweetField.ID, TweetField.CREATED_AT)
48+
.build());
6249

63-
// // Shutdown TweetsStreamListenersExecutor
64-
// try {
65-
// Thread.sleep(20000);
66-
// tsle.shutdown();
67-
// } catch (InterruptedException e) {
68-
// e.printStackTrace();
69-
// }
70-
71-
// // An example of how to use the streaming directly using the InputStream result (Without TweetsStreamListenersExecutor)
72-
// try{
73-
// JSON json = new JSON();
74-
// Type localVarReturnType = new TypeToken<StreamingTweet>(){}.getType();
75-
// BufferedReader reader = new BufferedReader(new InputStreamReader(streamResult));
76-
// String line = reader.readLine();
77-
// while (line != null) {
78-
// if(line.isEmpty()) {
79-
// System.err.println("==> " + line.isEmpty());
80-
// line = reader.readLine();
81-
// continue;
82-
// }
83-
// Object jsonObject = json.getGson().fromJson(line, localVarReturnType);
84-
// System.out.println(jsonObject != null ? jsonObject.toString() : "Null object");
85-
// line = reader.readLine();
86-
// }
87-
// }catch (Exception e) {
88-
// e.printStackTrace();
89-
// System.out.println(e);
90-
// }
91-
} catch (ApiException e) {
92-
System.err.println("Status code: " + e.getCode());
93-
System.err.println("Reason: " + e.getResponseBody());
94-
System.err.println("Response headers: " + e.getResponseHeaders());
95-
e.printStackTrace();
96-
}
9750
}
9851
}
9952

100-
class Responder implements com.twitter.clientlib.TweetsStreamListener {
53+
class Responder implements TweetsStreamListener {
10154
@Override
10255
public void actionOnTweetsStream(StreamingTweet streamingTweet) {
10356
if(streamingTweet == null) {

examples/src/main/java/com/twitter/clientlib/TweetsStreamListenersExecutor.java renamed to src/main/java/com/twitter/clientlib/stream/TweetsStreamExecutor.java

Lines changed: 61 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -20,60 +20,70 @@
2020
*/
2121

2222

23-
package com.twitter.clientlib;
23+
package com.twitter.clientlib.stream;
2424

2525

2626
import java.io.BufferedReader;
2727
import java.io.InputStream;
2828
import java.io.InputStreamReader;
29-
import java.lang.reflect.Type;
3029
import java.util.ArrayList;
31-
import java.util.LinkedList;
3230
import java.util.List;
33-
import java.util.Queue;
31+
import java.util.concurrent.BlockingQueue;
32+
import java.util.concurrent.ExecutorService;
33+
import java.util.concurrent.Executors;
34+
import java.util.concurrent.LinkedBlockingDeque;
35+
import java.util.stream.IntStream;
3436

35-
import com.google.gson.reflect.TypeToken;
37+
import com.fasterxml.jackson.core.JsonProcessingException;
38+
import com.fasterxml.jackson.databind.DeserializationFeature;
39+
import com.fasterxml.jackson.databind.JsonMappingException;
40+
import com.fasterxml.jackson.databind.ObjectMapper;
3641

3742
import com.twitter.clientlib.model.StreamingTweet;
38-
import com.twitter.clientlib.model.SingleTweetLookupResponse;
3943

40-
public class TweetsStreamListenersExecutor {
41-
private final ITweetsQueue tweetsQueue;
42-
private final List<TweetsStreamListener> listeners = new ArrayList<>();
43-
private final InputStream stream;
44+
public class TweetsStreamExecutor {
45+
private volatile BlockingQueue<String> rawTweets;
46+
private volatile BlockingQueue<StreamingTweet> tweets;
4447
private volatile boolean isRunning = true;
4548

46-
public TweetsStreamListenersExecutor(InputStream stream) {
47-
this.tweetsQueue = new LinkedListTweetsQueue();
48-
this.stream = stream;
49-
}
49+
private ExecutorService executorService;
50+
private final List<TweetsStreamListener> listeners = new ArrayList<>();
51+
private final InputStream stream;
5052

51-
public TweetsStreamListenersExecutor(ITweetsQueue tweetsQueue, InputStream stream) {
52-
this.tweetsQueue = tweetsQueue;
53+
public TweetsStreamExecutor(InputStream stream) {
54+
this.rawTweets = new LinkedBlockingDeque<>();
55+
this.tweets = new LinkedBlockingDeque<>();
5356
this.stream = stream;
5457
}
5558

5659
public void addListener(TweetsStreamListener toAdd) {
5760
listeners.add(toAdd);
5861
}
5962

60-
public void executeListeners() {
63+
public void removeListener(TweetsStreamListener toRemove) {
64+
listeners.remove(toRemove);
65+
}
66+
67+
public void start() {
6168
if (stream == null) {
6269
System.out.println("Error: stream is null.");
6370
return;
64-
} else if (this.tweetsQueue == null) {
65-
System.out.println("Error: tweetsQueue is null.");
66-
return;
6771
}
6872

69-
TweetsQueuer tweetsQueuer = new TweetsQueuer();
73+
RawTweetsQueuer rawTweetsQueuer = new RawTweetsQueuer();
7074
TweetsListenersExecutor tweetsListenersExecutor = new TweetsListenersExecutor();
75+
rawTweetsQueuer.start();
76+
int threads = 5; //TODO parametrize this
77+
executorService = Executors.newFixedThreadPool(threads);
78+
for (int i = 0; i < threads; i++) {
79+
executorService.submit(new ParseTweetsTask());
80+
}
7181
tweetsListenersExecutor.start();
72-
tweetsQueuer.start();
7382
}
7483

7584
public synchronized void shutdown() {
7685
isRunning = false;
86+
executorService.shutdown();
7787
System.out.println("TweetsStreamListenersExecutor is shutting down.");
7888
}
7989

@@ -87,7 +97,7 @@ private void processTweets() {
8797
StreamingTweet streamingTweet;
8898
try {
8999
while (isRunning) {
90-
streamingTweet = tweetsQueue.poll();
100+
streamingTweet = tweets.poll();
91101
if (streamingTweet == null) {
92102
Thread.sleep(100);
93103
continue;
@@ -102,15 +112,39 @@ private void processTweets() {
102112
}
103113
}
104114

105-
private class TweetsQueuer extends Thread {
115+
private class ParseTweetsTask implements Runnable {
116+
private final ObjectMapper objectMapper;
117+
private ParseTweetsTask() {
118+
this.objectMapper = new ObjectMapper();
119+
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
120+
}
121+
122+
@Override
123+
public void run() {
124+
while (isRunning) {
125+
try {
126+
String rawTweet = rawTweets.take();
127+
StreamingTweet tweet = objectMapper.readValue(rawTweet, StreamingTweet.class);
128+
tweets.put(tweet);
129+
} catch (InterruptedException e) {
130+
System.out.println("Fail 1");
131+
} catch (JsonMappingException e) {
132+
System.out.println("Fail 2");
133+
} catch (JsonProcessingException e) {
134+
System.out.println("Fail 3");
135+
}
136+
}
137+
}
138+
}
139+
140+
private class RawTweetsQueuer extends Thread {
141+
106142
@Override
107143
public void run() {
108144
queueTweets();
109145
}
110146

111147
public void queueTweets() {
112-
JSON json = new JSON();
113-
Type localVarReturnType = new TypeToken<SingleTweetLookupResponse>() {}.getType();
114148

115149
String line = null;
116150
try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream))) {
@@ -121,7 +155,7 @@ public void queueTweets() {
121155
continue;
122156
}
123157
try {
124-
tweetsQueue.add(StreamingTweet.fromJson(line));
158+
rawTweets.put(line);
125159
} catch (Exception interExcep) {
126160
interExcep.printStackTrace();
127161
}
@@ -134,21 +168,3 @@ public void queueTweets() {
134168
}
135169
}
136170

137-
interface ITweetsQueue {
138-
StreamingTweet poll();
139-
void add(StreamingTweet streamingTweet);
140-
}
141-
142-
class LinkedListTweetsQueue implements ITweetsQueue {
143-
private final Queue<StreamingTweet> tweetsQueue = new LinkedList<>();
144-
145-
@Override
146-
public StreamingTweet poll() {
147-
return tweetsQueue.poll();
148-
}
149-
150-
@Override
151-
public void add(StreamingTweet streamingTweet) {
152-
tweetsQueue.add(streamingTweet);
153-
}
154-
}

examples/src/main/java/com/twitter/clientlib/TweetsStreamListener.java renamed to src/main/java/com/twitter/clientlib/stream/TweetsStreamListener.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
*/
2121

2222

23-
package com.twitter.clientlib;
23+
package com.twitter.clientlib.stream;
2424

2525
import com.twitter.clientlib.model.StreamingTweet;
2626

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
package com.twitter.clientlib.stream;
2+
3+
import com.twitter.clientlib.ApiClient;
4+
import com.twitter.clientlib.ApiException;
5+
import com.twitter.clientlib.TwitterCredentialsBearer;
6+
import com.twitter.clientlib.api.TweetsApi;
7+
import com.twitter.clientlib.query.StreamQueryParameters;
8+
9+
import java.io.InputStream;
10+
import java.util.LinkedList;
11+
import java.util.List;
12+
13+
public class TwitterStream {
14+
private final TweetsApi tweets = new TweetsApi();
15+
16+
private final ApiClient apiClient = new ApiClient();
17+
18+
private final List<TweetsStreamListener> listeners = new LinkedList<>();
19+
20+
public TwitterStream() {
21+
initBasePath();
22+
tweets.setClient(apiClient);
23+
}
24+
25+
public void setTwitterCredentials(TwitterCredentialsBearer credentials) {
26+
apiClient.setTwitterCredentials(credentials);
27+
}
28+
29+
public void addListener(TweetsStreamListener listener) {
30+
listeners.add(listener);
31+
}
32+
33+
public void sampleStream(StreamQueryParameters streamParameters) {
34+
try {
35+
InputStream streamResult = tweets.sampleStream(streamParameters);
36+
TweetsStreamExecutor executor = new TweetsStreamExecutor(streamResult);
37+
listeners.forEach(executor::addListener);
38+
executor.start();
39+
} catch (ApiException e) {
40+
System.err.println("Status code: " + e.getCode());
41+
System.err.println("Reason: " + e.getResponseBody());
42+
System.err.println("Response headers: " + e.getResponseHeaders());
43+
e.printStackTrace();
44+
}
45+
}
46+
47+
private void initBasePath() {
48+
String basePath = System.getenv("TWITTER_API_BASE_PATH");
49+
apiClient.setBasePath(basePath != null ? basePath : "https://api.twitter.com");
50+
}
51+
52+
}

0 commit comments

Comments
 (0)