reduced duplication for closing executor service and switched to single thread deserialization layer

Author: kisaga

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

@@ -43,20 +43,21 @@ public static void main(String[] args) {
      * Check the 'security' tag of the required APIs in https://api.twitter.com/2/openapi.json in order
      * to use the right credential object.
      */
-    TwitterCredentialsBearer credentials = new TwitterCredentialsBearer("AAAAAAAAAAAAAAAAAAAAAJotdgEAAAAAmXZex2w6l1D6TcwC8KfripI3ADY%3DkKTjcNPxea1aAL8yYXJTdbAX1NkgEJii7SpXetxuP9GXXhSnOa");
+    TwitterCredentialsBearer credentials = new TwitterCredentialsBearer(System.getenv("TWITTER_BEARER_TOKEN"));
     TwitterStream twitterStream = new TwitterStream();
     twitterStream.setTwitterCredentials(credentials);
     twitterStream.addListener(new Responder());
+
 //    twitterStream.sampleStream(new StreamQueryParameters.Builder()
 //            .withTweetFields(TweetField.AUTHOR_ID, TweetField.ID, TweetField.CREATED_AT)
 //            .build());
 
     twitterStream.sampleStream(new StreamQueryParameters.Builder()
-                    .withTweetFields(TweetField.all().toArray(new TweetField[0]))
-                    .withMediaFields(MediaField.all().toArray(new MediaField[0]))
-                    .withUserFields(UserField.all().toArray(new UserField[0]))
-                    .withPollFields(PollField.all().toArray(new PollField[0]))
-                    .withPlaceFields(PlaceField.all().toArray(new PlaceField[0]))
+                    .withTweetFields(TweetField.all())
+                    .withMediaFields(MediaField.all())
+                    .withUserFields(UserField.all())
+                    .withPollFields(PollField.all())
+                    .withPlaceFields(PlaceField.all())
                     .build());
 
   }

src/main/java/com/twitter/clientlib/query/StreamQueryParameters.java

@@ -106,26 +106,51 @@ public Builder withTweetFields(TweetField... tweetFields) {
             return this;
         }
 
+        public Builder withTweetFields(List<TweetField> tweetFields) {
+            this.tweetFields.addAll(tweetFields);
+            return this;
+        }
+
         public Builder withMediaFields(MediaField... mediaFields) {
             this.mediaFields.addAll(Arrays.asList(mediaFields));
             return this;
         }
 
+        public Builder withMediaFields(List<MediaField> mediaFields) {
+            this.mediaFields.addAll(mediaFields);
+            return this;
+        }
+
         public Builder withPollFields(PollField... pollFields) {
             this.pollFields.addAll(Arrays.asList(pollFields));
             return this;
         }
 
+        public Builder withPollFields(List<PollField> pollFields) {
+            this.pollFields.addAll(pollFields);
+            return this;
+        }
+
         public Builder withUserFields(UserField... userFields) {
             this.userFields.addAll(Arrays.asList(userFields));
             return this;
         }
 
+        public Builder withUserFields(List<UserField> userFields) {
+            this.userFields.addAll(userFields);
+            return this;
+        }
+
         public Builder withPlaceFields(PlaceField... placeFields) {
             this.placeFields.addAll(Arrays.asList(placeFields));
             return this;
         }
 
+        public Builder withPlaceFields(List<PlaceField> placeFields) {
+            this.placeFields.addAll(placeFields);
+            return this;
+        }
+
         public Builder withBackfillMinutes(Integer backFillMinutes) {
             this.backFillMinutes = backFillMinutes;
             return this;

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

@@ -23,9 +23,6 @@
 package com.twitter.clientlib.stream;
 
 
-import java.io.BufferedReader;
-import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
@@ -50,8 +47,6 @@ public class TweetsStreamExecutor {
   private long startTime;
   private int tweetsCount = 0;
   private final int tweetsLimit = 250000;
-  private final int deserializationThreads = 1; //TODO parametrize this
-
   private ExecutorService rawTweetsQueuerService;
   private ExecutorService deserializationService;
   private ExecutorService listenersService;
@@ -83,45 +78,44 @@ public void start() {
     rawTweetsQueuerService = Executors.newSingleThreadExecutor();
     rawTweetsQueuerService.submit(new RawTweetsQueuer());
 
-    deserializationService = Executors.newFixedThreadPool(deserializationThreads);
-    for (int i = 0; i < deserializationThreads; i++) {
-      deserializationService.submit(new DeserializeTweetsTask());
-    }
+    deserializationService = Executors.newSingleThreadExecutor();
+    deserializationService.submit(new DeserializeTweetsTask());
 
     listenersService = Executors.newSingleThreadExecutor();
     listenersService.submit(new TweetsListenersTask());
   }
 
   public synchronized void shutdown() {
     isRunning = false;
-    rawTweetsQueuerService.shutdown();
-    deserializationService.shutdown();
-    listenersService.shutdown();
+    shutDownServices();
     try {
-      if (!rawTweetsQueuerService.awaitTermination(3, TimeUnit.SECONDS)) {
-        rawTweetsQueuerService.shutdownNow();
-        if (!rawTweetsQueuerService.awaitTermination(3, TimeUnit.SECONDS))
-          System.err.println("Pool did not terminate");
-      }
-      if (!deserializationService.awaitTermination(3, TimeUnit.SECONDS)) {
-        deserializationService.shutdownNow();
-        if (!deserializationService.awaitTermination(3, TimeUnit.SECONDS))
-          System.err.println("Pool did not terminate");
-      }
-      if (!listenersService.awaitTermination(3, TimeUnit.SECONDS)) {
-        listenersService.shutdownNow();
-        if (!listenersService.awaitTermination(3, TimeUnit.SECONDS))
-          System.err.println("Pool did not terminate");
-      }
+      terminateServices();
     } catch (InterruptedException ie) {
-      rawTweetsQueuerService.shutdown();
-      deserializationService.shutdown();
-      listenersService.shutdown();
+      shutDownServices();
       Thread.currentThread().interrupt();
     }
     System.out.println("TweetsStreamListenersExecutor is shutting down.");
   }
 
+  private void shutDownServices() {
+    rawTweetsQueuerService.shutdown();
+    deserializationService.shutdown();
+    listenersService.shutdown();
+  }
+
+  private void terminateServices() throws InterruptedException {
+    terminateService(rawTweetsQueuerService);
+    terminateService(deserializationService);
+    terminateService(listenersService);
+  }
+  private void terminateService(ExecutorService executorService) throws InterruptedException {
+    if (!executorService.awaitTermination(3, TimeUnit.SECONDS)) {
+      executorService.shutdownNow();
+      if (!executorService.awaitTermination(3, TimeUnit.SECONDS))
+        System.err.println("Pool did not terminate");
+    }
+  }
+
   private class RawTweetsQueuer implements Runnable {
 
     @Override
@@ -130,7 +124,6 @@ public void run() {
     }
 
     public void queueTweets() {
-
       String line = null;
       try {
         while (isRunning) {