Rx Java mergeDelayError không hoạt động như mong đợi

Question

Tôi đang sử dụng RxJava và ứng dụng Android với RxAndroid. Tôi đang sử dụng mergeDelayError để kết hợp hai cuộc gọi mạng phù hợp với retro thành một trong những quan sát có thể xử lý các mục được phát ra nếu phát ra một và lỗi nếu có. Điều này không hoạt động và nó chỉ kích hoạt hành vi onError khi gặp phải lỗi. Bây giờ để kiểm tra điều này, tôi đã chuyển sang một ví dụ rất đơn giản và vẫn là successAction không bao giờ được gọi khi tôi có một cuộc gọi onError. Xem ví dụ bên dưới.

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

Hành động thành công sẽ chỉ được gọi nếu tôi sử dụng hai lần quan sát thành công. Tôi có thiếu một cái gì đó với cách mergeDelayError có nghĩa vụ phải làm việc?

EDIT:

tôi đã tìm thấy rằng nếu tôi loại bỏ các observeOn và subscribeOn mọi thứ hoạt động như mong đợi. Tôi cần phải xác định chủ đề và nghĩ rằng đó là toàn bộ điểm của việc sử dụng Rx. Bất kỳ ý tưởng tại sao chỉ định những Schedulers sẽ phá vỡ hành vi?

Answer 1

Điều này vẫn có vẻ giống như một lỗi trong toán tử mergeDelayError nhưng tôi có thể làm cho nó hoạt động bằng cách sao chép người quan sát và Đăng ký cho từng quan sát được.

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

Tôi nghĩ rằng bạn không chờ đợi cho sự kiện thiết bị đầu cuối và các chủ đề chính bỏ trước khi sự kiện này được chuyển giao cho người quan sát của bạn. Các thử nghiệm sau đi cho tôi với RxJava 1.0.14:

@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

Sử dụng .observeOn (AndroidSchedulers.mainThread(), true) thay vì .observeOn (AndroidSchedulers.mainThread()

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

Trên đây là chữ ký của chức năng observeOn. Sau công trình mã.

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) { 

        } 
       }); 

Got thủ thuật này từ C oncatDelayError thread: https://github.com/ReactiveX/RxJava/issues/3908#issuecomment-217999009