本文整理了Java中io.reactivex.Observable.replay()方法的一些代码示例，展示了Observable.replay()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Observable.replay()方法的具体详情如下：
包路径：io.reactivex.Observable
类名称：Observable
方法名：replay

Observable.replay介绍

[英]Returns a ConnectableObservable that shares a single subscription to the underlying ObservableSource that will replay all of its items and notifications to any future Observer. A Connectable ObservableSource resembles an ordinary ObservableSource, except that it does not begin emitting items when it is subscribed to, but only when its connect method is called.

Scheduler: This version of replay does not operate by default on a particular Scheduler.
[中]返回一个ConnectableObservable，它共享对基础ObserveSource的单个订阅，该订阅将向任何未来的观察者重播其所有项目和通知。可连接的ObserveSource类似于普通ObserveSource，不同之处在于它在订阅时不会开始发送项，而只有在调用其connect方法时才会发送项。
调度程序：默认情况下，此版本的replay不会在特定调度程序上运行。

代码示例

代码示例来源：origin: ReactiveX/RxJava

@Override
  public ConnectableObservable<T> call() {
    return parent.replay(bufferSize);
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Override
  public ConnectableObservable<T> call() {
    return parent.replay();
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Override
  public ConnectableObservable<T> call() {
    return parent.replay(bufferSize, time, unit, scheduler);
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void replayBoundedSelectorNull() {
  just1.replay((Function<Observable<Integer>, Observable<Integer>>)null, 1, 1, TimeUnit.SECONDS);
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void replayTimeBoundedSchedulerNull() {
  just1.replay(1, TimeUnit.SECONDS, null);
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void replaySelectorTimeBoundedSchedulerNull() {
  just1.replay(new Function<Observable<Integer>, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Observable<Integer> v) {
      return v;
    }
  }, 1, TimeUnit.SECONDS, null);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testUnsubscribeSource() throws Exception {
  Action unsubscribe = mock(Action.class);
  Observable<Integer> o = Observable.just(1).doOnDispose(unsubscribe).replay().autoConnect();
  o.subscribe();
  o.subscribe();
  o.subscribe();
  verify(unsubscribe, never()).run();
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void replaySelectorReturnsNull() {
  just1.replay(new Function<Observable<Integer>, Observable<Object>>() {
    @Override
    public Observable<Object> apply(Observable<Integer> o) {
      return null;
    }
  }).blockingSubscribe();
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void replaySelectorTimeBoundedUnitNull() {
  just1.replay(new Function<Observable<Integer>, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Observable<Integer> v) {
      return v;
    }
  }, 1, null, Schedulers.single());
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void replayBoundedSelectorReturnsNull() {
  just1.replay(new Function<Observable<Integer>, Observable<Object>>() {
    @Override
    public Observable<Object> apply(Observable<Integer> v) {
      return null;
    }
  }, 1, 1, TimeUnit.SECONDS).blockingSubscribe();
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replaySelectorReturnsNull() {
  Observable.just(1)
  .replay(new Function<Observable<Integer>, Observable<Object>>() {
    @Override
    public Observable<Object> apply(Observable<Integer> v) throws Exception {
      return null;
    }
  })
  .test()
  .assertFailureAndMessage(NullPointerException.class, "The selector returned a null ObservableSource");
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replaySelectorReturnsNullScheduled() {
  Observable.just(1)
  .replay(new Function<Observable<Integer>, Observable<Object>>() {
    @Override
    public Observable<Object> apply(Observable<Integer> v) throws Exception {
      return null;
    }
  }, Schedulers.trampoline())
  .test()
  .assertFailureAndMessage(NullPointerException.class, "The selector returned a null ObservableSource");
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replayTime() {
  Observable.just(1).replay(1, TimeUnit.MINUTES)
  .autoConnect()
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replaySizeAndTime() {
  Observable.just(1).replay(1, 1, TimeUnit.MILLISECONDS)
  .autoConnect()
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void source() {
  Observable<Integer> source = Observable.range(1, 3);
  assertSame(source, (((HasUpstreamObservableSource<?>)source.replay())).source());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replaySizeScheduler() {
  Observable.just(1).replay(1, Schedulers.computation())
  .autoConnect()
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replayMaxInt() {
  Observable.range(1, 2)
  .replay(Integer.MAX_VALUE)
  .autoConnect()
  .test()
  .assertResult(1, 2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void cancelOnArrival() {
  Observable.range(1, 2)
  .replay(Integer.MAX_VALUE)
  .autoConnect()
  .test(true)
  .assertEmpty();
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void testTake() {
    TestObserver<Integer> to = new TestObserver<Integer>();

    Observable<Integer> cached = Observable.range(1, 100).replay().autoConnect();
    cached.take(10).subscribe(to);

    to.assertNoErrors();
    to.assertTerminated();
    to.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
    // FIXME no longer assertable
//        ts.assertUnsubscribed();
  }

代码示例来源：origin: ReactiveX/RxJava

@Test
public void replayIsUnsubscribed() {
  ConnectableObservable<Integer> co = Observable.just(1).concatWith(Observable.<Integer>never())
  .replay();
  if (co instanceof Disposable) {
    assertTrue(((Disposable)co).isDisposed());
    Disposable connection = co.connect();
    assertFalse(((Disposable)co).isDisposed());
    connection.dispose();
    assertTrue(((Disposable)co).isDisposed());
  }
}