/**
 * Copyright 2016 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 io.reactivex.schedulers;

import static org.junit.Assert.*;

import java.util.*;
import java.util.concurrent.*;

import org.junit.Test;

import io.reactivex.Scheduler.Worker;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.schedulers.Schedulers;

public class SchedulerLifecycleTest {
    @Test
    public void testShutdown() throws InterruptedException {
        tryOutSchedulers();
        
        System.out.println("testShutdown >> Giving time threads to spin-up");
        Thread.sleep(500);
        
        Set<Thread> rxThreads = new HashSet<Thread>();
        for (Thread t : Thread.getAllStackTraces().keySet()) {
            if (t.getName().startsWith("Rx")) {
                rxThreads.add(t);
            }
        }
        Schedulers.shutdown();
        System.out.println("testShutdown >> Giving time to threads to stop");
        Thread.sleep(500);
        
        StringBuilder b = new StringBuilder();
        for (Thread t : rxThreads) {
            if (t.isAlive()) {
                b.append("Thread " + t + " failed to shutdown\r\n");
                for (StackTraceElement ste : t.getStackTrace()) {
                    b.append("  ").append(ste).append("\r\n");
                }
            }
        }
        if (b.length() > 0) {
            System.out.print(b);
            System.out.println("testShutdown >> Restarting schedulers...");
            Schedulers.start(); // restart them anyways
            fail("Rx Threads were still alive:\r\n" + b);
        }
        
        System.out.println("testShutdown >> Restarting schedulers...");
        Schedulers.start();
        
        tryOutSchedulers();
    }

    private void tryOutSchedulers() throws InterruptedException {
        final CountDownLatch cdl = new CountDownLatch(4);
        
        final Runnable countAction = new Runnable() {
            @Override
            public void run() {
                cdl.countDown();
            }
        };
        
        CompositeDisposable csub = new CompositeDisposable();
        
        try {
            Worker w1 = Schedulers.computation().createWorker();
            csub.add(w1);
            w1.schedule(countAction);
            
            Worker w2 = Schedulers.io().createWorker();
            csub.add(w2);
            w2.schedule(countAction);
            
            Worker w3 = Schedulers.newThread().createWorker();
            csub.add(w3);
            w3.schedule(countAction);
            
            Worker w4 = Schedulers.single().createWorker();
            csub.add(w4);
            w4.schedule(countAction);
            
            
            if (!cdl.await(3, TimeUnit.SECONDS)) {
                fail("countAction was not run by every worker");
            }
        } finally {
            csub.dispose();
        }
    }
    
    @Test
    public void testStartIdempotence() throws InterruptedException {
        tryOutSchedulers();
        
        System.out.println("testStartIdempotence >> giving some time");
        Thread.sleep(500);
        
        Set<Thread> rxThreads = new HashSet<Thread>();
        for (Thread t : Thread.getAllStackTraces().keySet()) {
            if (t.getName().startsWith("Rx")) {
                rxThreads.add(t);
                System.out.println("testStartIdempotence >> " + t);
            }
        }
        System.out.println("testStartIdempotence >> trying to start again");
        Schedulers.start();
        System.out.println("testStartIdempotence >> giving some time again");
        Thread.sleep(500);
        
        Set<Thread> rxThreads2 = new HashSet<Thread>();
        for (Thread t : Thread.getAllStackTraces().keySet()) {
            if (t.getName().startsWith("Rx")) {
                rxThreads2.add(t);
                System.out.println("testStartIdempotence >>>> " + t);
            }
        }
        
        assertEquals(rxThreads, rxThreads2);
    }
}
@@ -720,20 +720,17 @@
 g.junit.
-Test
+*
 ;%0A%0Aimpor
@@ -3661,32 +3661,38 @@
 hread%3E rxThreads
+Before
  = new HashSet%3CT
@@ -3829,32 +3829,38 @@
        rxThreads
+Before
 .add(t);%0A       
@@ -4192,17 +4192,21 @@
 xThreads
-2
+After
  = new H
@@ -4359,17 +4359,21 @@
 xThreads
-2
+After
 .add(t);
@@ -4463,32 +4463,33 @@
 %7D%0A        %7D%0A
+%0A
         
 %0A        ass
@@ -4480,30 +4480,183 @@
     
-%0A        assertEquals(
+// cached threads may get dropped between the two checks%0A        rxThreadsAfter.removeAll(rxThreadsBefore);%0A        %0A        Assert.assertTrue(%22Some new threads appeared: %22 + 
 rxTh
@@ -4660,16 +4660,21 @@
 xThreads
+After
 , rxThre
@@ -4676,17 +4676,31 @@
 xThreads
-2
+After.isEmpty()
 );%0A    %7D

fix SchedulerLifecycle.testStartIdempotence vs cached threads dying