Added TwitterClient.stopFilteredStream() variant with a timeout. (#406)

src/main/java/io/github/redouane59/twitter/ITwitterClientV2.java

@@ -23,6 +23,7 @@
 import io.github.redouane59.twitter.dto.user.UserList;
 import java.util.List;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 public interface ITwitterClientV2 {
@@ -172,10 +173,20 @@ public interface ITwitterClientV2 {
    */
   Future<Response> startFilteredStream(IAPIEventListener listener, int backfillMinutes);
 
+  /**
+   * Stops the filtered stream with the result of the startFilteredStream. It'll wait a maximum of timeout
+   * before giving up and returning false.  If timeout isn't hit, it'll close the socket opened.
+   *
+   * @param responseFuture Future<Response> given by startFilteredStream
+   * @param timeout long How long to wait
+   * @param unit TimeUnit Units for timeout
+   */
+  boolean stopFilteredStream(Future<Response> response, long timeout, TimeUnit unit);
+
   /**
    * Stops the filtered stream with the result of the startFilteredStream. It'll close the socket opened.
    *
-   * @param response Future<Response> given by startFilteredStream
+   * @param responseFuture Future<Response> given by startFilteredStream
    */
   boolean stopFilteredStream(Future<Response> response);
 
@@ -641,4 +652,4 @@ public interface ITwitterClientV2 {
   boolean deleteTweet(String tweetId);
 
 
-}
+}

src/main/java/io/github/redouane59/twitter/TwitterClient.java

@@ -82,6 +82,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -1044,20 +1046,32 @@ public Future<Response> startFilteredStream(IAPIEventListener listener, int back
   }
 
   @Override
-  public boolean stopFilteredStream(Future<Response> response) {
+  public boolean stopFilteredStream(Future<Response> responseFuture, long timeout, TimeUnit unit) {
     try {
-      if (response.get() == null) {
+      Response response;
+      if (timeout > 0 && unit != null) {
+        response = responseFuture.get(timeout, unit);
+      } else {
+        response = responseFuture.get();
+      }
+
+      if (response == null) {
         return false;
       }
-      response.get().getStream().close();
+      response.getStream().close();
       return true;
-    } catch (IOException | InterruptedException | ExecutionException e) {
+    } catch (IOException | InterruptedException | ExecutionException | TimeoutException e) {
       LOGGER.error("Couldn't stopFilteredstream ", e);
       Thread.currentThread().interrupt();
     }
     return false;
   }
 
+  @Override
+  public boolean stopFilteredStream(Future<Response> responseFuture) {
+    return stopFilteredStream(responseFuture, 0, null);
+  }
+
   @Override
   public List<StreamRule> retrieveFilteredStreamRules() {
     String      url    = urlHelper.getFilteredStreamRulesUrl();