switched from InputStream and BufferedReader to okio BufferedSource

Author: kisaga

src/main/java/com/twitter/clientlib/ApiClient.java

@@ -29,6 +29,7 @@
 import okhttp3.logging.HttpLoggingInterceptor.Level;
 import okio.Buffer;
 import okio.BufferedSink;
+import okio.BufferedSource;
 import okio.Okio;
 import org.apache.oltu.oauth2.client.request.OAuthClientRequest.TokenRequestBuilder;
 import org.apache.oltu.oauth2.common.message.types.GrantType;
@@ -1099,11 +1100,10 @@ public <T> ApiResponse<T> execute(Call call, Type returnType) throws ApiExceptio
      * <p>Execute stream.</p>
      *
      * @param call a {@link okhttp3.Call} object
-     * @param returnType a {@link java.lang.reflect.Type} object
      * @return a {@link java.io.InputStream} object
      * @throws com.twitter.clientlib.ApiException if any.
      */
-    public InputStream executeStream(Call call, Type returnType) throws ApiException {
+    public BufferedSource executeStream(Call call) throws ApiException {
         try {
              Response response = call.execute();
              if (!response.isSuccessful()) {
@@ -1112,7 +1112,7 @@ public InputStream executeStream(Call call, Type returnType) throws ApiException
              if (response.body() == null) {
                  return null;
              }
-             return response.body().byteStream();
+             return response.body().source();
         } catch (IOException e) {
             throw new ApiException(e);
         }

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

@@ -78,6 +78,7 @@
 import java.io.InputStream;
 import javax.ws.rs.core.GenericType;
 
+import okio.BufferedSource;
 import org.apache.commons.lang3.StringUtils;
 
 public class TweetsApi extends ApiCommon {
@@ -1692,17 +1693,16 @@ private okhttp3.Call sampleStreamValidateBeforeCall(StreamQueryParameters stream
         <tr><td> 0 </td><td> The request has failed. </td><td>  -  </td></tr>
      </table>
      */
-    public InputStream sampleStream(StreamQueryParameters streamParameters) throws ApiException {
-        InputStream localVarResp = sampleStreamWithHttpInfo(streamParameters);
-        return localVarResp;
+    public BufferedSource sampleStream(StreamQueryParameters streamParameters) throws ApiException {
+        return sampleStreamWithHttpInfo(streamParameters);
     }
 
     /**
      * Calls the API using a retry mechanism to handle rate limits errors.
      *
      */
-    public InputStream sampleStream(Integer retries, StreamQueryParameters streamParameters) throws ApiException {
-        InputStream localVarResp;
+    public BufferedSource sampleStream(Integer retries, StreamQueryParameters streamParameters) throws ApiException {
+        BufferedSource localVarResp;
         try{
           localVarResp = sampleStream(streamParameters);
         }
@@ -1730,11 +1730,10 @@ public InputStream sampleStream(Integer retries, StreamQueryParameters streamPar
         <tr><td> 0 </td><td> The request has failed. </td><td>  -  </td></tr>
      </table>
      */
-    public InputStream sampleStreamWithHttpInfo(StreamQueryParameters streamParameters) throws ApiException {
+    public BufferedSource sampleStreamWithHttpInfo(StreamQueryParameters streamParameters) throws ApiException {
         okhttp3.Call localVarCall = sampleStreamValidateBeforeCall(streamParameters, null);
         try {
-            Type localVarReturnType = new TypeToken<StreamingTweet>(){}.getType();
-            return localVarApiClient.executeStream(localVarCall, localVarReturnType);
+            return localVarApiClient.executeStream(localVarCall);
         } catch (ApiException e) {
             e.setErrorObject(localVarApiClient.getJSON().getGson().fromJson(e.getResponseBody(), new TypeToken<com.twitter.clientlib.model.ProblemOrError>(){}.getType()));
             throw e;
@@ -1857,17 +1856,16 @@ private okhttp3.Call searchStreamValidateBeforeCall(StreamQueryParameters stream
         <tr><td> 0 </td><td> The request has failed. </td><td>  -  </td></tr>
      </table>
      */
-    public InputStream searchStream(StreamQueryParameters streamParameters) throws ApiException {
-        InputStream localVarResp = searchStreamWithHttpInfo(streamParameters);
-        return localVarResp;
+    public BufferedSource searchStream(StreamQueryParameters streamParameters) throws ApiException {
+        return searchStreamWithHttpInfo(streamParameters);
     }
 
     /**
      * Calls the API using a retry mechanism to handle rate limits errors.
      *
      */
-    public InputStream searchStream(Integer retries, StreamQueryParameters streamParameters) throws ApiException {
-        InputStream localVarResp;
+    public BufferedSource searchStream(Integer retries, StreamQueryParameters streamParameters) throws ApiException {
+        BufferedSource localVarResp;
         try{
           localVarResp = searchStream(streamParameters);
         }
@@ -1895,11 +1893,10 @@ public InputStream searchStream(Integer retries, StreamQueryParameters streamPar
         <tr><td> 0 </td><td> The request has failed. </td><td>  -  </td></tr>
      </table>
      */
-    public InputStream searchStreamWithHttpInfo(StreamQueryParameters streamParameters) throws ApiException {
+    public BufferedSource searchStreamWithHttpInfo(StreamQueryParameters streamParameters) throws ApiException {
         okhttp3.Call localVarCall = searchStreamValidateBeforeCall(streamParameters, null);
         try {
-            Type localVarReturnType = new TypeToken<FilteredStreamingTweet>(){}.getType();
-            return localVarApiClient.executeStream(localVarCall, localVarReturnType);
+            return localVarApiClient.executeStream(localVarCall);
         } catch (ApiException e) {
             e.setErrorObject(localVarApiClient.getJSON().getGson().fromJson(e.getResponseBody(), new TypeToken<com.twitter.clientlib.model.ProblemOrError>(){}.getType()));
             throw e;

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

@@ -40,6 +40,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import com.twitter.clientlib.model.StreamingTweet;
+import okio.BufferedSource;
 
 public class TweetsStreamExecutor {
   private volatile BlockingQueue<String> rawTweets;
@@ -56,9 +57,9 @@ public class TweetsStreamExecutor {
   private ExecutorService listenersService;
 
   private final List<TweetsStreamListener> listeners = new ArrayList<>();
-  private InputStream stream;
+  private BufferedSource stream;
 
-  public TweetsStreamExecutor(InputStream stream) {
+  public TweetsStreamExecutor(BufferedSource stream) {
     this.rawTweets = new LinkedBlockingDeque<>();
     this.tweets = new LinkedBlockingDeque<>();
     this.stream = stream;
@@ -131,9 +132,9 @@ public void run() {
     public void queueTweets() {
 
       String line = null;
-      try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream))) {
+      try {
         while (isRunning) {
-          line = reader.readLine();
+          line = stream.readUtf8Line();
           if(line == null || line.isEmpty()) {
             continue;
           }

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

@@ -5,6 +5,7 @@
 import com.twitter.clientlib.TwitterCredentialsBearer;
 import com.twitter.clientlib.api.TweetsApi;
 import com.twitter.clientlib.query.StreamQueryParameters;
+import okio.BufferedSource;
 
 import java.io.InputStream;
 import java.util.LinkedList;
@@ -32,7 +33,7 @@ public void addListener(TweetsStreamListener listener) {
 
     public void sampleStream(StreamQueryParameters streamParameters) {
         try {
-            InputStream streamResult = tweets.sampleStream(streamParameters);
+            BufferedSource streamResult = tweets.sampleStream(streamParameters);
             TweetsStreamExecutor executor = new TweetsStreamExecutor(streamResult);
             listeners.forEach(executor::addListener);
             executor.start();