/**
 * Copyright 2013 Netflix, Inc.
 * 
 * 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 rx.operators;

import static org.junit.Assert.*;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import rx.Observable;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.util.functions.Action1;
import rx.util.functions.Func1;

public class OperationParallelMergeTest {

    @Test
    public void testParallelMerge() {
        PublishSubject<String> p1 = PublishSubject.<String> create();
        PublishSubject<String> p2 = PublishSubject.<String> create();
        PublishSubject<String> p3 = PublishSubject.<String> create();
        PublishSubject<String> p4 = PublishSubject.<String> create();

        Observable<Observable<String>> fourStreams = Observable.<Observable<String>> from(p1, p2, p3, p4);

        Observable<Observable<String>> twoStreams = OperationParallelMerge.parallelMerge(fourStreams, 2);
        Observable<Observable<String>> threeStreams = OperationParallelMerge.parallelMerge(fourStreams, 3);

        List<? super Observable<String>> fourList = fourStreams.toList().toBlockingObservable().last();
        List<? super Observable<String>> threeList = threeStreams.toList().toBlockingObservable().last();
        List<? super Observable<String>> twoList = twoStreams.toList().toBlockingObservable().last();

        System.out.println("two list: " + twoList);
        System.out.println("three list: " + threeList);
        System.out.println("four list: " + fourList);

        assertEquals(4, fourList.size());
        assertEquals(3, threeList.size());
        assertEquals(2, twoList.size());
    }

    @Test
    public void testNumberOfThreads() {
        final ConcurrentHashMap<String, String> threads = new ConcurrentHashMap<String, String>();
        Observable.merge(getStreams())
                .toBlockingObservable().forEach(new Action1<String>() {

                    @Override
                    public void call(String o) {
                        System.out.println("o: " + o + " Thread: " + Thread.currentThread());
                        threads.put(Thread.currentThread().getName(), Thread.currentThread().getName());
                    }
                });

        // without injecting anything, the getStream() method uses Interval which runs on a default scheduler
        assertEquals(Runtime.getRuntime().availableProcessors(), threads.keySet().size());

        // clear
        threads.clear();

        // now we parallelMerge into 3 streams and observeOn for each
        // we expect 3 threads in the output
        OperationParallelMerge.parallelMerge(getStreams(), 3)
                .flatMap(new Func1<Observable<String>, Observable<String>>() {

                    @Override
                    public Observable<String> call(Observable<String> o) {
                        // for each of the parallel 
                        return o.observeOn(Schedulers.newThread());
                    }
                })
                .toBlockingObservable().forEach(new Action1<String>() {

                    @Override
                    public void call(String o) {
                        System.out.println("o: " + o + " Thread: " + Thread.currentThread());
                        threads.put(Thread.currentThread().getName(), Thread.currentThread().getName());
                    }
                });

        assertEquals(3, threads.keySet().size());
    }

    @Test
    public void testNumberOfThreadsOnScheduledMerge() {
        final ConcurrentHashMap<String, String> threads = new ConcurrentHashMap<String, String>();

        // now we parallelMerge into 3 streams and observeOn for each
        // we expect 3 threads in the output
        Observable.merge(OperationParallelMerge.parallelMerge(getStreams(), 3, Schedulers.newThread()))
                .toBlockingObservable().forEach(new Action1<String>() {

                    @Override
                    public void call(String o) {
                        System.out.println("o: " + o + " Thread: " + Thread.currentThread());
                        threads.put(Thread.currentThread().getName(), Thread.currentThread().getName());
                    }
                });

        assertEquals(3, threads.keySet().size());
    }

    private static Observable<Observable<String>> getStreams() {
        return Observable.range(0, 10).map(new Func1<Integer, Observable<String>>() {

            @Override
            public Observable<String> call(final Integer i) {
                return Observable.interval(10, TimeUnit.MILLISECONDS).map(new Func1<Long, String>() {

                    @Override
                    public String call(Long l) {
                        return "Stream " + i + "  Value: " + l;
                    }
                }).take(5);
            }
        });
    }
}

@@ -2320,36 +2320,32 @@
 rentHashMap%3C
-String, Stri
+Long, Lo
 ng%3E threads 
@@ -2372,722 +2372,33 @@
 Map%3C
-String, String%3E();%0A        Observable.merge(getStreams())%0A                .toBlockingObservable().forEach(new Action1%3CString%3E() %7B%0A%0A                    @Override%0A                    public void call(String o) %7B%0A                        System.out.println(%22o: %22 + o + %22 Thread: %22 + Thread.currentThread());%0A                        threads.put(Thread.currentThread().getName(), Thread.currentThread().getName());%0A                    %7D%0A                %7D);%0A%0A        // without injecting anything, the getStream() method uses Interval which runs on a default scheduler%0A        assertEquals(Runtime.getRuntime().availableProcessors(), threads.keySet().size());%0A%0A        // clear%0A        threads.clear();%0A%0A        // now we
+Long, Long%3E();%0A        //
  par
@@ -3135,32 +3135,40 @@
 .currentThread()
+.getId()
 );%0A             
@@ -3208,36 +3208,34 @@
 rentThread().get
-Name
+Id
 (), Thread.curre
@@ -3240,36 +3240,34 @@
 rentThread().get
-Name
+Id
 ());%0A           
@@ -3453,28 +3453,24 @@
 HashMap%3C
-String, Stri
+Long, Lo
 ng%3E thre
@@ -3501,20 +3501,16 @@
 Map%3C
-String, Stri
+Long, Lo
 ng%3E(
@@ -3967,32 +3967,40 @@
 .currentThread()
+.getId()
 );%0A             
@@ -4048,20 +4048,18 @@
 ad().get
-Name
+Id
 (), Thre
@@ -4084,12 +4084,10 @@
 .get
-Name
+Id
 ());

Fix non-deterministic unit test