Rx Java mergeDelayError beklendiği gibi çalışmıyor

Question

RxAndroid ile RxJava ve Android uygulamasında kullanıyorum. İki retro uygun ağ görüşmesini bir gözlenebilir olarak birleştirmek için mergeDelayError kullanıyorum, ya biri yandığında yayılan öğeleri işleyecek ve ya bir tane varsa, hata. Bu çalışmıyor ve yalnızca bir hatayla karşılaştığında onError eylemini başlatıyor. Şimdi bunu test etmek için çok basit bir örneğe geçtim ve hala bir onError çağrısı olduğunda successAction hiçbir zaman çağrılmıyor. Aşağıdaki örneğe bakın.

Observable.mergeDelayError(
       Observable.error(new RuntimeException()), 
       Observable.just("Hello") 
      ) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribeOn(Schedulers.io()) 
      .finallyDo(completeAction) 
      .subscribe(successAction, errorAction); 

Başarı eylemi, yalnızca iki başarı gözlenebilirse kullanırsam çağrılır. MergeDelayError'un nasıl çalışması gerektiğini bir şeyle özlüyorum mu?

DÜZENLEME: Ben kaldırırsanız beklendiği gibi observeOn ve subscribeOn her şey çalışıyor olduğunu tespit ettik

. Rx kullanmanın tüm noktası olan konuları ve düşünceyi belirtmem gerekiyor. Bu Schedulers'u neden belirtmek, davranışı bozar?

Answer 1

Bu hala mergeDelayError operatöründe bir hata gibi görünüyor, ancak gözlemlenebilir olanları gözlemlemek için Gözlemciyi çoğaltma ve Abone olmayla çalışmayı başarabiliyordum. etkinlikler gözlemci teslim edilmeden önce seni terminali olay ve ana iş parçacığı beklemeyin düşünüyorum

Observable.mergeDelayError(
      Observable.error(new RuntimeException()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeOn(Schedulers.io()), 
      Observable.just("Hello") 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeOn(Schedulers.io()) 
     ) 
     .finallyDo(completeAction) 
     .subscribe(successAction, errorAction); 

Answer 2

kapanıyor. Aşağıdaki test RxJava 1.0.14 benim için geçer:

@Test 
public void errorDelayed() { 
    TestSubscriber<Object> ts = TestSubscriber.create(); 
    Observable.mergeDelayError(
      Observable.error(new RuntimeException()), 
      Observable.just("Hello") 
     ) 
     .subscribeOn(Schedulers.io()).subscribe(ts); 

    ts.awaitTerminalEvent(); 

    ts.assertError(RuntimeException.class); 
    ts.assertValue("Hello"); 
} 

Answer 3

Kullanım .observeOn (AndroidSchedulers.mainThread(), true) yerine .observeOn (AndroidSchedulers.mainThread()

public final Observable<T> observeOn(Scheduler scheduler, boolean delayError) { 
     return observeOn(scheduler, delayError, RxRingBuffer.SIZE); 
    } 

Üstü observeOn fonksiyonunun imzadır. kod çalışmaları takiben.

Observable.mergeDelayError(
       Observable.error(new RuntimeException()), 
       Observable.just("Hello") 
     ) 
       .observeOn(AndroidSchedulers.mainThread(), true) 
       .subscribeOn(Schedulers.io()) 
       .subscribe(new Subscriber<String>() { 
        @Override 
        public void onCompleted() { 

        } 

        @Override 
        public void onError(Throwable e) { 

        } 

        @Override 
        public void onNext(String s) { 

        } 
       }); 

C'den bu hileyi var oncatDelayError konu: https://github.com/ReactiveX/RxJava/issues/3908#issuecomment-217999009