checking permorfance for different number of threads

Author: kisaga

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

@@ -25,7 +25,11 @@
 
 import com.twitter.clientlib.model.*;
 import com.twitter.clientlib.query.StreamQueryParameters;
+import com.twitter.clientlib.query.model.MediaField;
+import com.twitter.clientlib.query.model.PlaceField;
+import com.twitter.clientlib.query.model.PollField;
 import com.twitter.clientlib.query.model.TweetField;
+import com.twitter.clientlib.query.model.UserField;
 import com.twitter.clientlib.stream.TweetsStreamListener;
 import com.twitter.clientlib.stream.TwitterStream;
 
@@ -39,20 +43,28 @@ 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(System.getenv("TWITTER_BEARER_TOKEN"));
+    TwitterCredentialsBearer credentials = new TwitterCredentialsBearer("AAAAAAAAAAAAAAAAAAAAAJotdgEAAAAAmXZex2w6l1D6TcwC8KfripI3ADY%3DkKTjcNPxea1aAL8yYXJTdbAX1NkgEJii7SpXetxuP9GXXhSnOa");
     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.AUTHOR_ID, TweetField.ID, TweetField.CREATED_AT)
-            .build());
+                    .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]))
+                    .build());
 
   }
 }
 
 class Responder implements TweetsStreamListener {
   @Override
-  public void actionOnTweetsStream(StreamingTweet streamingTweet) {
+  public synchronized void actionOnTweetsStream(StreamingTweet streamingTweet) {
     if(streamingTweet == null) {
       System.err.println("Error: actionOnTweetsStream - streamingTweet is null ");
       return;

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

@@ -32,21 +32,30 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
-import java.util.stream.IntStream;
+import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import com.twitter.clientlib.JSON;
 import com.twitter.clientlib.model.StreamingTweet;
 
 public class TweetsStreamExecutor {
   private volatile BlockingQueue<String> rawTweets;
   private volatile BlockingQueue<StreamingTweet> tweets;
   private volatile boolean isRunning = true;
 
-  private ExecutorService executorService;
+  private long startTime;
+  private int tweetsCount = 0;
+  private final int tweetsLimit = 80000;
+  private final int threads = 1; //TODO parametrize this
+
+  private ExecutorService rawTweetsQueuerService;
+  private ExecutorService deserializationService;
+  private ExecutorService listenersService;
+
   private final List<TweetsStreamListener> listeners = new ArrayList<>();
   private final InputStream stream;
 
@@ -69,51 +78,83 @@ public void start() {
       System.out.println("Error: stream is null.");
       return;
     }
+    startTime = System.currentTimeMillis();
 
-    RawTweetsQueuer rawTweetsQueuer = new RawTweetsQueuer();
-    TweetsListenersExecutor tweetsListenersExecutor = new TweetsListenersExecutor();
-    rawTweetsQueuer.start();
-    int threads = 5; //TODO parametrize this
-    executorService = Executors.newFixedThreadPool(threads);
+    rawTweetsQueuerService = Executors.newSingleThreadExecutor();
+    rawTweetsQueuerService.submit(new RawTweetsQueuer());
+
+    deserializationService = Executors.newFixedThreadPool(threads);
     for (int i = 0; i < threads; i++) {
-      executorService.submit(new ParseTweetsTask());
+      deserializationService.submit(new ParseTweetsTask());
     }
-    tweetsListenersExecutor.start();
+
+    listenersService = Executors.newSingleThreadExecutor();
+    listenersService.submit(new TweetsListenersTask());
   }
 
   public synchronized void shutdown() {
     isRunning = false;
-    executorService.shutdown();
+    rawTweetsQueuerService.shutdown();
+    deserializationService.shutdown();
+    listenersService.shutdown();
+    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");
+      }
+    } catch (InterruptedException ie) {
+      rawTweetsQueuerService.shutdown();
+      deserializationService.shutdown();
+      listenersService.shutdown();
+      Thread.currentThread().interrupt();
+    }
     System.out.println("TweetsStreamListenersExecutor is shutting down.");
   }
 
-  private class TweetsListenersExecutor extends Thread {
+  private class RawTweetsQueuer implements Runnable {
+
     @Override
     public void run() {
-      processTweets();
+      queueTweets();
     }
 
-    private void processTweets() {
-      StreamingTweet streamingTweet;
-      try {
+    public void queueTweets() {
+
+      String line = null;
+      try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream))) {
         while (isRunning) {
-          streamingTweet = tweets.poll();
-          if (streamingTweet == null) {
-            Thread.sleep(100);
+          line = reader.readLine();
+          if(line == null || line.isEmpty()) {
+            Thread.sleep(10);
             continue;
           }
-          for (TweetsStreamListener listener : listeners) {
-            listener.actionOnTweetsStream(streamingTweet);
+          try {
+            rawTweets.put(line);
+          } catch (Exception interExcep) {
+            interExcep.printStackTrace();
           }
         }
       } catch (Exception e) {
         e.printStackTrace();
+        shutdown();
       }
     }
   }
 
   private class ParseTweetsTask implements Runnable {
     private final ObjectMapper objectMapper;
+
     private ParseTweetsTask() {
       this.objectMapper = new ObjectMapper();
       objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
@@ -137,32 +178,35 @@ public void run() {
     }
   }
 
-  private class RawTweetsQueuer extends Thread {
-
+  private class TweetsListenersTask implements Runnable {
     @Override
     public void run() {
-      queueTweets();
+      processTweets();
     }
 
-    public void queueTweets() {
-
-      String line = null;
-      try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream))) {
+    private void processTweets() {
+      StreamingTweet streamingTweet;
+      try {
         while (isRunning) {
-          line = reader.readLine();
-          if(line == null || line.isEmpty()) {
-            Thread.sleep(100);
+          streamingTweet = tweets.poll();
+          if (streamingTweet == null) {
+            Thread.sleep(10);
             continue;
           }
-          try {
-            rawTweets.put(line);
-          } catch (Exception interExcep) {
-            interExcep.printStackTrace();
+          for (TweetsStreamListener listener : listeners) {
+            listener.actionOnTweetsStream(streamingTweet);
+          }
+          tweetsCount++;
+          if(tweetsCount == tweetsLimit) {
+            long stopTime = System.currentTimeMillis();
+            long durationInMillis = stopTime - startTime;
+            double seconds = durationInMillis / 1000.0;
+            System.out.println("Total duration in seconds: " + seconds);
+            shutdown();
           }
         }
       } catch (Exception e) {
         e.printStackTrace();
-        shutdown();
       }
     }
   }

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

@@ -45,7 +45,7 @@ public void sampleStream(StreamQueryParameters streamParameters) {
     }
 
     private void initBasePath() {
-        String basePath = System.getenv("TWITTER_API_BASE_PATH");
+        String basePath = "http://localhost:8080";
         apiClient.setBasePath(basePath != null ? basePath : "https://api.twitter.com");
     }