第一个元素之前的RxJava Observable超时

Question

我有设备发送给我ping，我使用observable为此。但在第一次ping之前我们开始连接并且需要一些时间。因此我想先ping有10秒超时。我是这样做的：

public Observable<Ping> getPing() {
    ConnectableObservable<Ping> observable = device.connectToDevice().publish();

    Observable<Ping> firstWithTimeout = observable.take(1).timeout(10, TimeUnit.SECONDS);
    Observable<Ping> fromSecondWithoutTimeout = observable.skip(1);

    Observable<Ping> mergedObservable = firstWithTimeout.mergeWith(fromSecondWithoutTimeout)
            .doOnDispose(() -> disconnect(bluetoothDevice))
            .doOnError(error -> disconnect(bluetoothDevice));

    observable.connect();
    return mergedObservable;
}

我试用的测试

Subject<Ping> observable = PublishSubject.create();
when(device.connect()).thenReturn(observable);
TestObserver<Ping> testSubscriber = TestObserver.create();
getPing.subscribe(testSubscriber);
observable.onNext(new Ping());

testSubscriber.assertValueCount(1);

此测试将失败，因为TimeoutException，尽管我立即发送ping。

Answer 1

请看一下这个设置：

JUnit5 / RxJava2

我认为你错误的模拟错误配置

当（device.connect（））thenReturn（观察到的）。

请看看我的实施情况。在使用每个方法调用创建新的observable时，无需使用发布/连接。在设备中使用autoConnect for method-impl connectToDevice（）

  Device device;

  @BeforeEach
  void setUp() {
    device = mock(Device.class);
  }

  @Test
  void name() throws Exception {
    Subject<Ping> observable = PublishSubject.create();

    when(device.connectToDevice()).thenReturn(observable);

    TestObserver<Ping> test = getPing(Schedulers.computation()).test();
    observable.onNext(new Ping());

    test.assertValueCount(1);
  }

  @Test
  void name2() throws Exception {
    Subject<Ping> observable = PublishSubject.create();

    when(device.connectToDevice()).thenReturn(observable);

    TestScheduler testScheduler = new TestScheduler();
    TestObserver<Ping> test = getPing(testScheduler).test();

    testScheduler.advanceTimeBy(20, TimeUnit.SECONDS);

    observable.onNext(new Ping());

    test.assertError(TimeoutException.class);
  }

  private Observable<Ping> getPing(Scheduler scheduler) {

    return device
        .connectToDevice()
        .take(1)
        .timeout(10, TimeUnit.SECONDS, scheduler)
        .doOnDispose(() -> disconnect())
        .doOnError(error -> disconnect());
  }

  private void disconnect() {}

  interface Device {
    Observable<Ping> connectToDevice();
  }

  class Ping {}

Answer 2

有一个超载的timeout运算符，非常适合这里：

timeout(ObservableSource<U> firstTimeoutIndicator, Function<? super T, ? extends ObservableSource<V>> itemTimeoutIndicator)

假设您的可观察参考是testObserable，您只需执行以下操作：

testObservable.timeout(
        Observable.timer(5L, TimeUnit.SECONDS), // here you set first item timeout
        ignored -> Observable.never() // for there elements there is no time function
)