<commit_before>/*
 * Copyright (c) 2008-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.cometd.common;

import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.IntFunction;

import org.cometd.bayeux.Promise;

/**
 * <p>Processes asynchronously a sequence of elements, producing a result that
 * is the reduction of the results produced by the processing of each element.</p>
 * <p>This class implements the asynchronous version of the following synchronous
 * {@code for} loop, but where the processing of each element may be asynchronous.</p>
 * <pre>
 * R result;
 * for (T element : list) {
 *     (result, proceed) = operation.apply(result, element);
 *     if (!proceed) {
 *         break;
 *     }
 * }
 * </pre>
 * <p>Using this class, the loop above becomes:</p>
 * <pre>
 * R zero;
 * AsyncFoldLeft.run(list, zero, (result, element, loop) -&gt; {
 *     CompletableFuture&lt;R&gt; future = processAsync(element);
 *     future.whenComplete((r, x) -&gt; {
 *         if (x == null) {
 *             R reduced = reduce(result, r);
 *             if (shouldIterate(r)) {
 *                 loop.proceed(reduced);
 *             } else {
 *                 loop.leave(reduced);
 *             }
 *         } else {
 *             loop.fail(x);
 *         }
 *     })
 * }, Promise.complete((r, x) -&gt; {
 *     // Process final result or failure.
 * });
 * </pre>
 */
public class AsyncFoldLeft {
    public static <T, R> void run(T[] array, R zero, Operation<T, R> operation, Promise<R> promise) {
        int size = array.length;
        if (size == 0) {
            promise.succeed(zero);
        } else {
            IndexedLoop<T, R> loop = new IndexedLoop<>(i -> array[i], size, zero, operation, promise);
            loop.run();
        }
    }

    /**
     * <p>Processes the given {@code list} of elements.</p>
     * <p>The initial result {@code zero} is returned if the list is empty.</p>
     * <p>For each element the {@link Operation#apply(Object, Object, Loop) operation}
     * function is invoked.</p>
     *
     * @param list      the elements to process
     * @param zero      the initial result
     * @param operation the operation to invoke for each element
     * @param promise   the promise to notify of the final result
     * @param <T>       the type of element
     * @param <R>       the type of the result
     */
    public static <T, R> void run(List<T> list, R zero, Operation<T, R> operation, Promise<R> promise) {
        int size = list.size();
        if (size == 0) {
            promise.succeed(zero);
        } else {
            IndexedLoop<T, R> loop = new IndexedLoop<>(list::get, size, zero, operation, promise);
            loop.run();
        }
    }

    /**
     * <p>Processes the given {@code collection} of elements.</p>
     * <p>The initial result {@code zero} is returned if the collection is empty.</p>
     * <p>For each element the {@link Operation#apply(Object, Object, Loop) operation}
     * function is invoked.</p>
     *
     * @param collection the elements to process
     * @param zero       the initial result
     * @param operation  the operation to invoke for each element
     * @param promise    the promise to notify of the final result
     * @param <T>        the type of element
     * @param <R>        the type of the result<commit_msg>Fixes #899 - AsyncFoldLeft not threadsafe if collection is concurrently modified can result in closed Websockets<commit_after>/*
 * Copyright (c) 2008-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.cometd.common;

import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.IntFunction;

import org.cometd.bayeux.Promise;

/**
 * <p>Processes asynchronously a sequence of elements, producing a result that
 * is the reduction of the results produced by the processing of each element.</p>
 * <p>This class implements the asynchronous version of the following synchronous
 * {@code for} loop, but where the processing of each element may be asynchronous.</p>
 * <pre>
 * R result;
 * for (T element : list) {
 *     (result, proceed) = operation.apply(result, element);
 *     if (!proceed) {
 *         break;
 *     }
 * }
 * </pre>
 * <p>Using this class, the loop above becomes:</p>
 * <pre>
 * R zero;
 * AsyncFoldLeft.run(list, zero, (result, element, loop) -&gt; {
 *     CompletableFuture&lt;R&gt; future = processAsync(element);
 *     future.whenComplete((r, x) -&gt; {
 *         if (x == null) {
 *             R reduced = reduce(result, r);
 *             if (shouldIterate(r)) {
 *                 loop.proceed(reduced);
 *             } else {
 *                 loop.leave(reduced);
 *             }
 *         } else {
 *             loop.fail(x);
 *         }
 *     })
 * }, Promise.complete((r, x) -&gt; {
 *     // Process final result or failure.
 * });
 * </pre>
 */
public class AsyncFoldLeft {
    public static <T, R> void run(T[] array, R zero, Operation<T, R> operation, Promise<R> promise) {
        int size = array.length;
        if (size == 0) {
            promise.succeed(zero);
        } else {
            IndexedLoop<T, R> loop = new IndexedLoop<>(i -> array[i], size, zero, operation, promise);
            loop.run();
        }
    }

    /**
     * <p>Processes the given {@code list} of elements.</p>
     * <p>The initial result {@code zero} is returned if the list is empty.</p>
     * <p>For each element the {@link Operation#apply(Object, Object, Loop) operation}
     * function is invoked.</p>
     *
     * @param list      the elements to process
     * @param zero      the initial result
     * @param operation the operation to invoke for each element
     * @param promise   the promise to notify of the final result
     * @param <T>       the type of element
     * @param <R>       the type of the result
     * @deprecated use {@link #run(Collection, Object, Operation, Promise)} instead.
     */
    @Deprecated
    public static <T, R> void run(List<T> list, R zero, Operation<T, R> operation, Promise<R> promise) {
        run((Collection<T>)list, zero, operation, promise);
    }

    /**
     * <p>Processes the given {@code collection} of elements.</p>
     * <p>The initial result {@code zero} is returned if the collection is empty.</p>
     * <p>For each element the {@link Operation#apply(Object, Object, Loop) operation}
     * function is invoked.</p>
     * <p>The {@code collection} should have a "stable" iterator, i.e. it should not be
     * affected if the collection is modified concurrently by another thread, or by the
     * same thread in th {@code operation} during the iteration.</p>
     *
     * @param collection the elements to process
     * @param zero       the initial result
     * @param operation  the operation to invoke for each element
     * @param promise    the promise to notify of the final result
     * @param <T>        the type of element
     * @param <R>        the type of the result
{'fraction_non_alphanumeric': 0.11486242941247146, 'fraction_numerical': 0.0036044695422323683, 'mean_word_length': 2.5848406546080964, 'pattern_counts': {'":': 0, '<': 78, '<?xml version=': 0, '>': 80, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4558788', 'n_tokens_mistral': 2379, 'n_tokens_neox': 2333, 'n_words': 893}