RxTest.java

   private static Logger logger = Logger.getLogger(RxTest.class);

    private rx.Scheduler tickScheduler;

    private ExecutorService queueExecutor;

    private rx.Scheduler queueScheduler;

    @BeforeClass
    public void initClass() {
        tickScheduler = Schedulers.test();
        ExecutorService tickExecutor = Executors.newSingleThreadExecutor(
                new ThreadFactoryBuilder().setNameFormat("job-queue-pool-%d").build());
        tickScheduler = Schedulers.from(tickExecutor);
        queueExecutor = Executors.newSingleThreadExecutor(
                new ThreadFactoryBuilder().setNameFormat("job-queue-pool-%d").build());
        queueScheduler = Schedulers.from(queueExecutor);
    }
    
    private static class Iteration {
        public int iteration;
        public long timeout;

        public Iteration(int iteration, long timeout) {
            this.iteration = iteration;
            this.timeout = timeout;
        }
    }

    @Test
    public void runJobs() throws Exception {
        CountDownLatch latch = new CountDownLatch(3);
        Queue<Iteration> iterations = new LinkedList<>();
        iterations.offer(new Iteration(0, 5000));
        iterations.offer(new Iteration(1, 2500));
        iterations.offer(new Iteration(2, 500));

        doOnTick(() -> {
            int iteration = iterations.peek().iteration;
            long timeout = iterations.peek().timeout;
            iterations.poll();
            Observable.just(iteration).map(i ->
                    Completable.fromAction(() -> {
                        logger.debug("Starting long job " + iteration);
                        sleep(timeout);
                    }).concatWith(Completable.fromAction(() -> {
                        logger.debug("Starting short job " + iteration);
                        sleep(100);
                    })).subscribeOn(Schedulers.io())
            )
                    .flatMap(completable -> Observable.just(completable).subscribeOn(Schedulers.computation()))
                    .observeOn(Schedulers.computation())
                    .reduce(Completable::merge)
                    .flatMap(completable -> Completable.fromAction(completable::await).toObservable())
                    .subscribe(
                            next -> {
                            },
                            t -> logger.warn("There was an error", t),
                            () -> {
                                logger.debug("Finished iteration " + iteration);
                                latch.countDown();
                            }
                    );
        });

        assertTrue(latch.await(20, TimeUnit.SECONDS));
    }

    private void sleep(long timeout) {
        try {
            Thread.sleep(timeout);
        } catch (InterruptedException e) {
            logger.info("Interrupted", e);
        }
    }

    private void doOnTick(Action0 action) {
        Observable.interval(0, 1, TimeUnit.SECONDS, tickScheduler)
                .takeUntil(d -> !running)
                .observeOn(queueScheduler)
                .subscribe(tick -> action.call(), t -> logger.warn(t));
    }
    
}