code: public static <T, X extends Throwable> Tuple3<CompletableFuture<Subscription>, Runnable, CompletableFuture<Boolean>> forEachX(
              final FutureStream<T> stream, final long x, final Consumer<? super T> consumerElement) {
          final CompletableFuture<Subscription> subscription = new CompletableFuture<>();
          final CompletableFuture<Boolean> streamCompleted = new CompletableFuture<>();
          return tuple(subscription, () -> {
              stream.subscribe(new Subscriber<T>() {
  
                  @Override
                  public void onSubscribe(final Subscription s) {
                      Objects.requireNonNull(s);
                      if(x!=0)
                          s.request(x);
                      subscription.complete(s);
                  }
  
                  @Override
                  public void onNext(final T t) {
                      consumerElement.accept(t);
  
                  }
  
                  @Override
                  public void onError(final Throwable t) {
  
                  }
  
                  @Override
                  public void onComplete() {
                      streamCompleted.complete(true);
  
                  }
  
              });
          } , streamCompleted);
      }
docs: Perform a forEach operation over the Stream, without closing it, consuming only the specified number of elements from
  the Stream, at this time. More elements can be consumed later, by called request on the returned Subscription
  
  <pre>
  @{code
  Subscription next = StreamUtils.forEach(Stream.of(1,2,3,4),2,System.out::println);
  
  System.out.println("First batch processed!");
  
  next.request(2);
  
  System.out.println("Second batch processed!");
  
  //prints
  1
  2
  First batch processed!
  3
  4
  Second batch processed!
  }
  </pre>
  
  @param stream - the Stream to consume data from
  @param x To consume from the Stream at this time
  @param consumerElement To accept incoming events from the Stream
  @return Subscription so that further processing can be continued or cancelled.