Запись синхронных unittests для Rx v2 Flowable

Question

Я конвертирую свой проект из Rx v1 в Rx v2, и сейчас я в процессе изменения некоторых из v1 Observable s до v2 Flowable s.

(Его в проекте Android с UnitTests, написанной на Groovy, используя Спки)

Обычно я бы просто переопределить Планировщик с помощью крючка. Я все еще могу сделать это в v2, зарегистрировав обработчики планировщика. Это делает синхронным Observable s, всегда используя (новый?) Schedulers.single(). Однако Flowable s по-прежнему являются асинхронными из-за механики обратного давления (?).

Я попытался решения, что с помощью следующих действий:

Flowable<LogEntry> flowable = Flowable.create(new FlowableOnSubscribe<LogEntry>() { 
    @Override 
    void subscribe(FlowableEmitter<LogEntry> emitter) throws Exception { 
     for (def log : logs) { 
      emitter.onNext(log) 
     } 

     emitter.onComplete() 
    } 
}, FlowableEmitter.BackpressureMode.NONE); 

Но до сих пор делает их асинхронно.

Я перекрываться планировщиками, как это:

RxJavaPlugins.reset() 
RxJavaPlugins.setIoSchedulerHandler(new Function<Scheduler, Scheduler>() { 
    @Override 
    Scheduler apply(Scheduler scheduler) throws Exception { 
     return Schedulers.single() 
    } 
}) 

RxAndroidPlugins.reset() 
RxAndroidPlugins.setMainThreadSchedulerHandler(new Function<Scheduler, Scheduler>() { 
    @Override 
    Scheduler apply(Scheduler scheduler) throws Exception { 
     return Schedulers.from(new Executor() { 
      @Override 
      void execute(Runnable command) { 
       command.run() 
      } 
     }) 
    } 
}) 

Я не могу показаться, чтобы выяснить, почему Observable сек ведут себя синхронным, как это, но Flowable s нет (рядом с противодавлением механиком)

Answer 1

Schedulers.single() - однопоточный асинхронный планировщик. Вам нужно Schedulers.trampoline() остаться в той же теме.