"Yeniden yürütme" ve "autoConnect" kullanıldığında istisna daha fazla istisna "

Question

Yalnızca yeni değer bir öncekinden farklı olduğunda, göze çarpan sıcak gözlemlenebilir değerlerin (bundan önce -1 ile başlayan) bir gözlemlenebilir oluşturmak istiyorum. . Ayrıca en yeni değerin hemen yeni abonelere yayılmasını istiyorum. Aşağıdaki kod ile geldim:

PublishSubject<Integer> hotObservable = PublishSubject.create(); 

Observable<Integer> observable = hotObservable 
     .startWith(-1) 
     .distinctUntilChanged() 
     .replay(1) 
     .autoConnect(0); 

Ancak bu başarısız (bakılmaksızın observable abone önce yayar neyi hotObservable her zaman -1) Bilmiyorum zaman İlginçtir java.lang.IllegalStateException: more produced than requested ile yeni aboneye yayılan ilk değerden sonra otomatik olarak bağlanır, ancak abone olunan aboneler düzgün çalışır, son değeri alır ve sonra güncelleştirir:

.

Çalışmak için replay(1).autoConnect(0) alamıyorum ve bir şeyi özlüyorum gibi hissediyorum; neden abone olmak ve abonelikten çıkmak isteseydim, autoConnect(0) ise? Böyle gözlemlenebilir oluşturmanın uygun yolu nedir?

Observable<Integer> observable = hotObservable 
     .startWith(-1) 
     .distinctUntilChanged() 
     .replay(1) 
     .autoConnect(); // With (0) it fails 

observable.subscribe().unsubscribe(); // Needed if we don't auto connnect 

hotObservable.onNext(1); 
hotObservable.onNext(2); 
hotObservable.onNext(3); // I want this value to be received by new subscriber 

TestSubscriber<Integer> subscriber = TestSubscriber.create(); 
observable.subscribe(subscriber); 

subscriber.assertNoErrors(); 
subscriber.assertValues(3); 

Answer 1

Ben RxJava 1.1.3 yukarıdaki kod ile More produced than requested hata alamadım:

Burada autoConnect(); observable.subscribe().unsubscribe() kullanmadığınız sürece başarısız test yöntemi bu. onaylama işlemi başarısız

nedeni onun Aboneler herhangi aslında talep dek replay memba bir şey istemeyeceğiz olmasıdır. TestSubscriber abone ilk ise, yayacak şekilde startWith tetikleyebilir -1 ve sonra başka bir şey almazlar dolayısıyla herhangi bir değeri korumaz PublishSubject geçer.

BehaviorSubject<Integer> hotObservable = BehaviorSubject.create(-1); 

Observable<Integer> observable = hotObservable.distinctUntilChanged(); 

hotObservable.onNext(1); 
hotObservable.onNext(2); 
hotObservable.onNext(3); 

TestSubscriber<Integer> subscriber = TestSubscriber.create(); 
observable.subscribe(subscriber); 

subscriber.assertNoErrors(); 
subscriber.assertValue(3);