/* (non-Javadoc)
@see com.oath.cyclops.types.Traversable#forEachAsync(org.reactivestreams.Subscriber)
969181
@Override
    public void subscribe(final Subscriber<? super T> s) {

       transformerStream().forEach(v->v.subscribe(s));

    }
1298134
public <T> void forEach(Stream<T> stream, Consumer<? super T> action) {
        Runnable runnable = () -> stream.forEach(action);
        ThreadMethods.forkJoinExecution(runnable, concurrencyConfiguration, stream.isParallel());
    }
686036
@Override
    public java.util.concurrent.Future<SetSubscriptionAttributesResult> setSubscriptionAttributesAsync(String subscriptionArn, String attributeName,
            String attributeValue, com.amazonaws.handlers.AsyncHandler<SetSubscriptionAttributesRequest, SetSubscriptionAttributesResult> asyncHandler) {

        return setSubscriptionAttributesAsync(new SetSubscriptionAttributesRequest().withSubscriptionArn(subscriptionArn).withAttributeName(attributeName)
                .withAttributeValue(attributeValue), asyncHandler);
    }
537068
@Override
    public ReactiveSeq<T> stream() {
        return ReactiveSeq.generate(() -> get())
                          .limit(1)
                          .filter(t -> t.isSuccess())
                          .map(t->t.orElse(null));
    }
992251
public Observable<Page<SubscriptionInner>> listAsync() {
        return listWithServiceResponseAsync()
            .map(new Func1<ServiceResponse<Page<SubscriptionInner>>, Page<SubscriptionInner>>() {
                @Override
                public Page<SubscriptionInner> call(ServiceResponse<Page<SubscriptionInner>> response) {
                    return response.body();
                }
            });
    }
43262
public ReactiveTask requestAsync(final long n) {
        return withSubscriptionAndTask(subscriptionAndTask.map2(c -> CompletableFuture.runAsync(() -> subscriptionAndTask._1().join()
                                                                                                                            .request(n),
                                                                                                exec)));
    }
939221
@Override
    public java.util.concurrent.Future<CreateStreamResult> createStreamAsync(String streamName, Integer shardCount,
            com.amazonaws.handlers.AsyncHandler<CreateStreamRequest, CreateStreamResult> asyncHandler) {

        return createStreamAsync(new CreateStreamRequest().withStreamName(streamName).withShardCount(shardCount), asyncHandler);
    }
1015311
public void forEach(final IntIntConsumer consumer)
    {
        final int[] entries = this.entries;
        final int initialValue = this.initialValue;
        @DoNotSub final int length = entries.length;

        for (@DoNotSub int i = 0; i < length; i += 2)
        {
            if (entries[i + 1] != initialValue) // lgtm [java/index-out-of-bounds]
            {
                consumer.accept(entries[i], entries[i + 1]); // lgtm [java/index-out-of-bounds]
            }
        }
    }
1290471
@Override
    public java.util.concurrent.Future<ListSAMLProvidersResult> listSAMLProvidersAsync(
            com.amazonaws.handlers.AsyncHandler<ListSAMLProvidersRequest, ListSAMLProvidersResult> asyncHandler) {

        return listSAMLProvidersAsync(new ListSAMLProvidersRequest(), asyncHandler);
    }
1027620
public static <K, V> void forEach(Map<K, V> map, KVConsumer<K, V> kvConsumer) {
		int index = 0;
		for (Entry<K, V> entry : map.entrySet()) {
			kvConsumer.accept(entry.getKey(), entry.getValue(), index);
			index++;
		}
	}
959189
@Override
    public java.util.concurrent.Future<GetShardIteratorResult> getShardIteratorAsync(String streamName, String shardId, String shardIteratorType,
            String startingSequenceNumber, com.amazonaws.handlers.AsyncHandler<GetShardIteratorRequest, GetShardIteratorResult> asyncHandler) {

        return getShardIteratorAsync(new GetShardIteratorRequest().withStreamName(streamName).withShardId(shardId).withShardIteratorType(shardIteratorType)
                .withStartingSequenceNumber(startingSequenceNumber), asyncHandler);
    }
138851
@Override
    public void forEach(Consumer<? super P_OUT> action) {
        evaluate(ForEachOps.makeRef(action, false));
    }
741982
@SafeVarargs
    public final <U> FutureStream<U> of(final U... array) {
        return fromStream(ReactiveSeq.of(array));
    }
1222325
@Override
    public java.util.concurrent.Future<ListSSHPublicKeysResult> listSSHPublicKeysAsync(
            com.amazonaws.handlers.AsyncHandler<ListSSHPublicKeysRequest, ListSSHPublicKeysResult> asyncHandler) {

        return listSSHPublicKeysAsync(new ListSSHPublicKeysRequest(), asyncHandler);
    }
983169
@SuppressWarnings("unchecked")
    public static <T> ReactiveSeq<T> takeUntil(final Stream<T> stream, final Predicate<? super T> predicate) {
        final Iterator<T> it = stream.iterator();

        class LimitUntil implements Iterator<T> {
            T next = (T) NULL;
            boolean test = false;

            void test() {
                if (!test && next == NULL && it.hasNext()) {
                    next = it.next();

                    if (test = predicate.test(next)) {
                        next = (T) NULL;
                        close(it); // need to close any open queues
                    }
                }
            }

            @Override
            public boolean hasNext() {
                test();
                return next != NULL;
            }

            @Override
            public T next() {
                if (next == NULL)
                    throw new NoSuchElementException();

                try {
                    return next;
                } finally {
                    next = (T) NULL;
                }
            }
        }

        return ReactiveSeq.fromIterator(new LimitUntil())
                  .onClose(() -> {
                      stream.close();
                  });
    }
122985
@Override
  public <R> StreamBuilderImpl<R,U>
  collect(SupplierSync<R> init,
          BiConsumerSync<R,T> accum,
          BiConsumerSync<R,R> combiner)
  {
    return new CollectSync<>(this, init, accum, combiner);
  }
963441
@SafeVarargs
    public static <T> void stream(Callback<T> callback, T... items) throws Exception
    {
        stream(callback, Arrays.asList(items));
    }