<commit_before>package org.apache.usergrid.persistence.collection.rx;


import java.util.ArrayList;
import java.util.List;

import rx.Observable;
import rx.Scheduler;
import rx.operators.OperationMerge;
import rx.util.functions.Func1;
import rx.util.functions.Func2;
import rx.util.functions.FuncN;
import rx.util.functions.Functions;


/**
 * A utility class that encapsulates many Funct1 operations that receive and emit the same type.
 * These functions are executed in parallel, then "zipped" into a single response.  This is useful
 * when you want to perform operations on a single initial observable, then combine the result into a
 * single observable to continue the sequence

 */
public class Concurrent<T, R> implements Func1<T, Observable<R>> {



    private final Func1<T, R>[] concurrent;
    private final Scheduler scheduler;

    private Concurrent(final Scheduler scheduler, final Func1<T, R>[] concurrent ){
        this.scheduler = scheduler;
        this.concurrent = concurrent;
    }

    @Override
      public Observable<R> call( final T input ) {

        List<Observable<R>> observables = new ArrayList<Observable<R>>();

        //put all our observables together for concurrency
        for( Func1<T, R> funct: concurrent){
            final Observable<R> observable = Observable.from(input).subscribeOn(scheduler).map( funct );

            observables.add( observable );
        }




        Observable.OnSubscribeFunc<R> merge = OperationMerge.merge( observables );
        Observable<R> newObservable = Observable.create( merge );


        //wait until the last operation completes to proceed
        return newObservable.takeLast( 1 );

      }


    /**
     * Create an instance of concurrent execution.  All functions specified in the list are invoked in parallel.
     * The results are then "zipped" into a single observable which is returned
     *
     *
     * @param scheduler The scheduler to be used to schedule the concurrent tasks
     * @param observable The observable we're invoking on
     * @param concurrent The concurrent operations we're invoking
     * @return
     */
    public static <T, R> Observable<R> concurrent( final Scheduler scheduler, final Observable<T> observable, final Func1<T, R>... concurrent ){
          return observable.mapMany(new Concurrent<T, R>(scheduler,  concurrent ));
    }
<commit_msg>Fixes bug with concurrent task wrapper<commit_after>package org.apache.usergrid.persistence.collection.rx;


import java.util.ArrayList;
import java.util.List;

import rx.Observable;
import rx.Scheduler;
import rx.operators.OperationMerge;
import rx.util.functions.Func1;
import rx.util.functions.Func2;
import rx.util.functions.FuncN;
import rx.util.functions.Functions;


/**
 * A utility class that encapsulates many Funct1 operations that receive and emit the same type.
 * These functions are executed in parallel, then "zipped" into a single response.  This is useful
 * when you want to perform operations on a single initial observable, then combine the result into a
 * single observable to continue the sequence

 */
public class Concurrent<T, R> implements Func1<T, Observable<R>> {



    private final Func1<T, R>[] concurrent;
    private final Scheduler scheduler;

    private Concurrent(final Scheduler scheduler, final Func1<T, R>[] concurrent ){
        this.scheduler = scheduler;
        this.concurrent = concurrent;
    }

    @Override
      public Observable<R> call( final T input ) {

        List<Observable<R>> observables = new ArrayList<Observable<R>>(concurrent.length);

        //put all our observables together for concurrency
        for( Func1<T, R> funct: concurrent){
            final Observable<R> observable = Observable.from(input).subscribeOn(scheduler).map( funct );

            observables.add( observable );
        }




        final Observable.OnSubscribeFunc<R> merge = OperationMerge.merge( observables );
        final Observable<R> newObservable = Observable.create( merge );


        //wait until the last operation completes to proceed
        return newObservable.takeLast( 1 );

      }


    /**
     * Create an instance of concurrent execution.  All functions specified in the list are invoked in parallel.
     * The results are then "zipped" into a single observable which is returned
     *
     *
     * @param scheduler The scheduler to be used to schedule the concurrent tasks
     * @param observable The observable we're invoking on
     * @param concurrent The concurrent operations we're invoking
     * @return
     */
    public static <T, R> Observable<R> concurrent( final Scheduler scheduler, final Observable<T> observable, final Func1<T, R>... concurrent ){
          return observable.mapMany(new Concurrent<T, R>(scheduler,  concurrent ));
    }

{'fraction_non_alphanumeric': 0.08513738551207327, 'fraction_numerical': 0.003746877601998335, 'mean_word_length': 3.3484162895927603, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '425551', 'n_tokens_mistral': 1328, 'n_tokens_neox': 1228, 'n_words': 459}