package xdean.jex.extra.rx2;

import static xdean.jex.util.log.LogUtil.debug;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory;
import java.util.concurrent.ForkJoinWorkerThread;
import java.util.concurrent.ThreadFactory;

import io.reactivex.Scheduler;
import io.reactivex.schedulers.Schedulers;
import xdean.jex.util.lang.FinalizeSupport;

public class RxSchedulers {
  public static Scheduler fixedSize(int size) {
    return autoClose(Executors.newFixedThreadPool(size, new ThreadFactory() {
      int i = 0;

      @Override
      public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setName("FixedSizeScheduler(size=" + size + ")-" + (++i));
        t.setPriority(Thread.NORM_PRIORITY);
        t.setDaemon(true);
        return t;
      }
    }));
  }

  public static Scheduler forkJoin(int size) {
    class FJWT extends ForkJoinWorkerThread {
      protected FJWT(ForkJoinPool pool) {
        super(pool);
      }
    }
    return autoClose(new ForkJoinPool(size, new ForkJoinWorkerThreadFactory() {
      int i = 0;

      @Override
      public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
        ForkJoinWorkerThread t = new FJWT(pool);
        t.setName("ForkJoinScheduler(parallelism=" + size + ")-" + (++i));
        t.setPriority(Thread.NORM_PRIORITY);
        t.setDaemon(true);
        return t;
      }
    }, null, true));
  }

  public static Scheduler autoClose(ExecutorService pool) {
    Scheduler scheduler = Schedulers.from(pool);
    FinalizeSupport.finalize(scheduler, () -> {
      debug().log("Shutdown the scheduler from: " + pool);
      pool.shutdown();
    });
    return scheduler;
  }
}

@@ -429,106 +429,530 @@
 vex.
-schedulers.Schedulers;%0D%0Aimport xdean.jex.util.lang.FinalizeSupport;%0D%0A%0D%0Apublic class RxSchedulers %7B
+internal.schedulers.RxThreadFactory;%0D%0Aimport io.reactivex.schedulers.Schedulers;%0D%0Aimport xdean.jex.extra.LazyValue;%0D%0Aimport xdean.jex.util.lang.FinalizeSupport;%0D%0A%0D%0Apublic class RxSchedulers %7B%0D%0A  /**%0D%0A   * @see https://github.com/ReactiveX/RxJava/issues/5822%0D%0A   */%0D%0A  private static final LazyValue%3CScheduler%3E NEW_IO = LazyValue.create(%0D%0A      () -%3E Schedulers.from(Executors.newCachedThreadPool(new RxThreadFactory(%22RxSchedulers.newIOScheduler%22))));%0D%0A%0D%0A  public static Scheduler newIO() %7B%0D%0A    return NEW_IO.get();%0D%0A  %7D%0D%0A
 %0D%0A  

add NEW_IO