/*
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * 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.rxjava4.internal.schedulers;

import static org.junit.Assert.*;

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

import org.junit.Test;

import io.reactivex.rxjava4.core.RxJavaTest;
import io.reactivex.rxjava4.core.Scheduler.Worker;
import io.reactivex.rxjava4.disposables.CompositeDisposable;
import io.reactivex.rxjava4.internal.schedulers.SchedulerMultiWorkerSupport.WorkerCallback;
import io.reactivex.rxjava4.schedulers.Schedulers;
import io.reactivex.rxjava4.testsupport.TestHelper;

public class SchedulerMultiWorkerSupportTest extends RxJavaTest {

    final int max = ComputationScheduler.MAX_THREADS;

    @Test
    public void moreThanMaxWorkers() {
        final List<Worker> list = new ArrayList<>();

        SchedulerMultiWorkerSupport mws = (SchedulerMultiWorkerSupport)Schedulers.computation();

        mws.createWorkers(max * 2, new WorkerCallback() {
            @Override
            public void onWorker(int i, Worker w) {
                list.add(w);
            }
        });

        assertEquals(max * 2, list.size());
    }

    @Test
    public void getShutdownWorkers() {
        final List<Worker> list = new ArrayList<>();

        ComputationScheduler.NONE.createWorkers(max * 2, new WorkerCallback() {
            @Override
            public void onWorker(int i, Worker w) {
                list.add(w);
            }
        });

        assertEquals(max * 2, list.size());

        for (Worker w : list) {
            assertEquals(ComputationScheduler.SHUTDOWN_WORKER, w);
        }
    }

    @Test
    public void distinctThreads() throws Exception {
        for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {

            @SuppressWarnings("resource")
            final CompositeDisposable composite = new CompositeDisposable();

            try {
                final CountDownLatch cdl = new CountDownLatch(max * 2);

                final Set<String> threads1 = Collections.synchronizedSet(new HashSet<>());

                final Set<String> threads2 = Collections.synchronizedSet(new HashSet<>());

                Runnable parallel1 = new Runnable() {
                    @Override
                    public void run() {
                        final List<Worker> list1 = new ArrayList<>();

                        SchedulerMultiWorkerSupport mws = (SchedulerMultiWorkerSupport)Schedulers.computation();

                        mws.createWorkers(max, new WorkerCallback() {
                            @Override
                            public void onWorker(int i, Worker w) {
                                list1.add(w);
                                composite.add(w);
                            }
                        });

                        Runnable run = new Runnable() {
                            @Override
                            public void run() {
                                threads1.add(Thread.currentThread().getName());
                                cdl.countDown();
                            }
                        };

                        for (Worker w : list1) {
                            w.schedule(run);
                        }
                    }
                };

                Runnable parallel2 = new Runnable() {
                    @Override
                    public void run() {
                        final List<Worker> list2 = new ArrayList<>();

                        SchedulerMultiWorkerSupport mws = (SchedulerMultiWorkerSupport)Schedulers.computation();

                        mws.createWorkers(max, new WorkerCallback() {
                            @Override
                            public void onWorker(int i, Worker w) {
                                list2.add(w);
                                composite.add(w);
                            }
                        });

                        Runnable run = new Runnable() {
                            @Override
                            public void run() {
                                threads2.add(Thread.currentThread().getName());
                                cdl.countDown();
                            }
                        };

                        for (Worker w : list2) {
                            w.schedule(run);
                        }
                    }
                };

                TestHelper.race(parallel1, parallel2);

                assertTrue(cdl.await(5, TimeUnit.SECONDS));

                assertEquals(threads1.toString(), max, threads1.size());
                assertEquals(threads2.toString(), max, threads2.size());
            } finally {
                composite.dispose();
            }
        }
    }
}
