Может ли кто-то понять java.lang.IllegalStateException: только один наблюдатель разрешен в этом кратком примере?

Question

Я понимаю, что 30491785 обсуждает это, но не было реального объяснения причин возникновения ошибки, а в примере использовался посторонний код. Я бы хотел задать вопрос с 5-строчным примером.

Вопрос в том, что происходит с gwindows, который не работает с swindows (кроме того, что типы разные), и есть ли какая-нибудь работа?

public class OneObservableError { 

    public static void main(String[] args) throws Exception { 
     Observable<Long> source = Observable.interval(1, TimeUnit.SECONDS); 
     Observable<Observable<Long>> swindows = source.window(source, s -> Observable.interval(3, TimeUnit.SECONDS)); 
     Observable<GroupedObservable<Long, Long>> groups = source.groupBy(x -> x % 4); 
     Observable<Observable<Observable<Long>>> gwindows 
       = groups.map(g -> g.window(g, i -> Observable.interval(3, TimeUnit.SECONDS))); 
     //swindows.flatMap(gw->gw).subscribe(System.out::println); //Works 
     gwindows.flatMap(gw -> gw).subscribe(System.out::println); //Fails with Only one Observable allowed  
     sleep(10000); 
    } 

Answer 1

GroupedObservable является одноадресным типом источника, и вы не можете использовать его более одного раза. Проблема заключается в вашем коде в g.window (g, ...), где оператор окна пытается дважды подписаться. Используйте публиковать (Function) для обмена Одноразовый г:

g.publish(gs -> gs.window(gs, ...).flatMap(gw -> gw))...