io.reactivex.Observable.ambArray()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(12.5k)|赞(0)|评价(0)|浏览(154)

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

Observable.ambArray介绍

[英]Mirrors the one ObservableSource in an array of several ObservableSources that first either emits an item or sends a termination notification.

Scheduler: ambArray does not operate by default on a particular Scheduler.
[中]镜像多个可观察资源数组中的一个可观察资源，这些可观察资源首先发出一个项目或发送终止通知。
调度程序：默认情况下，ambArray不会在特定调度程序上运行。

代码示例

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

  1. @Test(expected = NullPointerException.class)
  2. public void ambVarargsNull() {
  3.   Observable.ambArray((Observable<Object>[])null);
  4. }

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

  1. /**
  2.  * Mirrors the ObservableSource (current or provided) that first either emits an item or sends a termination
  3.  * notification.
  4.  * <p>
  5.  * <img width="640" height="385" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/ambWith.o.png" alt="">
  6.  * <dl>
  7.  *  <dt><b>Scheduler:</b></dt>
  8.  *  <dd>{@code ambWith} does not operate by default on a particular {@link Scheduler}.</dd>
  9.  * </dl>
  10.  *
  11.  * @param other
  12.  *            an ObservableSource competing to react first. A subscription to this provided source will occur after
  13.  *            subscribing to the current source.
  14.  * @return an Observable that emits the same sequence as whichever of the source ObservableSources first
  15.  *         emitted an item or sent a termination notification
  16.  * @see <a href="http://reactivex.io/documentation/operators/amb.html">ReactiveX operators documentation: Amb</a>
  17.  */
  18. @SuppressWarnings("unchecked")
  19. @CheckReturnValue
  20. @SchedulerSupport(SchedulerSupport.NONE)
  21. public final Observable<T> ambWith(ObservableSource<? extends T> other) {
  22.   ObjectHelper.requireNonNull(other, "other is null");
  23.   return ambArray(this, other);
  24. }

代码示例来源：origin: redisson/redisson

  1. /**
  2.  * Mirrors the ObservableSource (current or provided) that first either emits an item or sends a termination
  3.  * notification.
  4.  * <p>
  5.  * <img width="640" height="385" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/ambWith.o.png" alt="">
  6.  * <dl>
  7.  *  <dt><b>Scheduler:</b></dt>
  8.  *  <dd>{@code ambWith} does not operate by default on a particular {@link Scheduler}.</dd>
  9.  * </dl>
  10.  *
  11.  * @param other
  12.  *            an ObservableSource competing to react first. A subscription to this provided source will occur after
  13.  *            subscribing to the current source.
  14.  * @return an Observable that emits the same sequence as whichever of the source ObservableSources first
  15.  *         emitted an item or sent a termination notification
  16.  * @see <a href="http://reactivex.io/documentation/operators/amb.html">ReactiveX operators documentation: Amb</a>
  17.  */
  18. @SuppressWarnings("unchecked")
  19. @CheckReturnValue
  20. @SchedulerSupport(SchedulerSupport.NONE)
  21. public final Observable<T> ambWith(ObservableSource<? extends T> other) {
  22.   ObjectHelper.requireNonNull(other, "other is null");
  23.   return ambArray(this, other);
  24. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void noWinnerSuccessDispose() throws Exception {
  4.   for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
  5.     final AtomicBoolean interrupted = new AtomicBoolean();
  6.     final CountDownLatch cdl = new CountDownLatch(1);
  7.     Observable.ambArray(
  8.       Observable.just(1)
  9.         .subscribeOn(Schedulers.single())
  10.         .observeOn(Schedulers.computation()),
  11.       Observable.never()
  12.     )
  13.     .subscribe(new Consumer<Object>() {
  14.       @Override
  15.       public void accept(Object v) throws Exception {
  16.         interrupted.set(Thread.currentThread().isInterrupted());
  17.         cdl.countDown();
  18.       }
  19.     });
  20.     assertTrue(cdl.await(500, TimeUnit.SECONDS));
  21.     assertFalse("Interrupted!", interrupted.get());
  22.   }
  23. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void ambArrayEmpty() {
  4.   assertSame(Observable.empty(), Observable.ambArray());
  5. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void ambArrayOrder() {
  4.   Observable<Integer> error = Observable.error(new RuntimeException());
  5.   Observable.ambArray(Observable.just(1), error).test().assertValue(1).assertComplete();
  6. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test(expected = NullPointerException.class)
  3. public void ambVarargsOneIsNull() {
  4.   Observable.ambArray(Observable.never(), null).blockingLast();
  5. }

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

  1. @Test
  2. public void testAmb2() {
  3.   IOException expectedException = new IOException(
  4.       "fake exception");
  5.   Observable<String> observable1 = createObservable(new String[] {},
  6.       2000, new IOException("fake exception"));
  7.   Observable<String> observable2 = createObservable(new String[] {
  8.       "2", "22", "222", "2222" }, 1000, expectedException);
  9.   Observable<String> observable3 = createObservable(new String[] {},
  10.       3000, new IOException("fake exception"));
  11.   @SuppressWarnings("unchecked")
  12.   Observable<String> o = Observable.ambArray(observable1,
  13.       observable2, observable3);
  14.   Observer<String> observer = TestHelper.mockObserver();
  15.   o.subscribe(observer);
  16.   scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS);
  17.   InOrder inOrder = inOrder(observer);
  18.   inOrder.verify(observer, times(1)).onNext("2");
  19.   inOrder.verify(observer, times(1)).onNext("22");
  20.   inOrder.verify(observer, times(1)).onNext("222");
  21.   inOrder.verify(observer, times(1)).onNext("2222");
  22.   inOrder.verify(observer, times(1)).onError(expectedException);
  23.   inOrder.verifyNoMoreInteractions();
  24. }

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

  1. @Test
  2. public void testAmb() {
  3.   Observable<String> observable1 = createObservable(new String[] {
  4.       "1", "11", "111", "1111" }, 2000, null);
  5.   Observable<String> observable2 = createObservable(new String[] {
  6.       "2", "22", "222", "2222" }, 1000, null);
  7.   Observable<String> observable3 = createObservable(new String[] {
  8.       "3", "33", "333", "3333" }, 3000, null);
  9.   @SuppressWarnings("unchecked")
  10.   Observable<String> o = Observable.ambArray(observable1,
  11.       observable2, observable3);
  12.   Observer<String> observer = TestHelper.mockObserver();
  13.   o.subscribe(observer);
  14.   scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS);
  15.   InOrder inOrder = inOrder(observer);
  16.   inOrder.verify(observer, times(1)).onNext("2");
  17.   inOrder.verify(observer, times(1)).onNext("22");
  18.   inOrder.verify(observer, times(1)).onNext("222");
  19.   inOrder.verify(observer, times(1)).onNext("2222");
  20.   inOrder.verify(observer, times(1)).onComplete();
  21.   inOrder.verifyNoMoreInteractions();
  22. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void ambArraySingleElement() {
  4.   assertSame(Observable.never(), Observable.ambArray(Observable.never()));
  5. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void noWinnerCompleteDispose() throws Exception {
  4.   for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
  5.     final AtomicBoolean interrupted = new AtomicBoolean();
  6.     final CountDownLatch cdl = new CountDownLatch(1);
  7.     Observable.ambArray(
  8.       Observable.empty()
  9.         .subscribeOn(Schedulers.single())
  10.         .observeOn(Schedulers.computation()),
  11.       Observable.never()
  12.     )
  13.     .subscribe(Functions.emptyConsumer(), Functions.emptyConsumer(), new Action() {
  14.       @Override
  15.       public void run() throws Exception {
  16.         interrupted.set(Thread.currentThread().isInterrupted());
  17.         cdl.countDown();
  18.       }
  19.     });
  20.     assertTrue(cdl.await(500, TimeUnit.SECONDS));
  21.     assertFalse("Interrupted!", interrupted.get());
  22.   }
  23. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void disposed() {
  4.   TestHelper.checkDisposed(Observable.ambArray(Observable.never(), Observable.never()));
  5. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void noWinnerErrorDispose() throws Exception {
  4.   final TestException ex = new TestException();
  5.   for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
  6.     final AtomicBoolean interrupted = new AtomicBoolean();
  7.     final CountDownLatch cdl = new CountDownLatch(1);
  8.     Observable.ambArray(
  9.       Observable.error(ex)
  10.         .subscribeOn(Schedulers.single())
  11.         .observeOn(Schedulers.computation()),
  12.       Observable.never()
  13.     )
  14.     .subscribe(Functions.emptyConsumer(), new Consumer<Throwable>() {
  15.       @Override
  16.       public void accept(Throwable e) throws Exception {
  17.         interrupted.set(Thread.currentThread().isInterrupted());
  18.         cdl.countDown();
  19.       }
  20.     });
  21.     assertTrue(cdl.await(500, TimeUnit.SECONDS));
  22.     assertFalse("Interrupted!", interrupted.get());
  23.   }
  24. }

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

  1. @Test
  2. public void testAmb3() {
  3.   Observable<String> observable1 = createObservable(new String[] {
  4.       "1" }, 2000, null);
  5.   Observable<String> observable2 = createObservable(new String[] {},
  6.       1000, null);
  7.   Observable<String> observable3 = createObservable(new String[] {
  8.       "3" }, 3000, null);
  9.   @SuppressWarnings("unchecked")
  10.   Observable<String> o = Observable.ambArray(observable1,
  11.       observable2, observable3);
  12.   Observer<String> observer = TestHelper.mockObserver();
  13.   o.subscribe(observer);
  14.   scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS);
  15.   InOrder inOrder = inOrder(observer);
  16.   inOrder.verify(observer, times(1)).onComplete();
  17.   inOrder.verifyNoMoreInteractions();
  18. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void testSubscriptionOnlyHappensOnce() throws InterruptedException {
  4.   final AtomicLong count = new AtomicLong();
  5.   Consumer<Disposable> incrementer = new Consumer<Disposable>() {
  6.     @Override
  7.     public void accept(Disposable d) {
  8.       count.incrementAndGet();
  9.     }
  10.   };
  11.   //this aync stream should emit first
  12.   Observable<Integer> o1 = Observable.just(1).doOnSubscribe(incrementer)
  13.       .delay(100, TimeUnit.MILLISECONDS).subscribeOn(Schedulers.computation());
  14.   //this stream emits second
  15.   Observable<Integer> o2 = Observable.just(1).doOnSubscribe(incrementer)
  16.       .delay(100, TimeUnit.MILLISECONDS).subscribeOn(Schedulers.computation());
  17.   TestObserver<Integer> to = new TestObserver<Integer>();
  18.   Observable.ambArray(o1, o2).subscribe(to);
  19.   to.awaitTerminalEvent(5, TimeUnit.SECONDS);
  20.   to.assertNoErrors();
  21.   assertEquals(2, count.get());
  22. }

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

  1. @Test
  2. public void onCompleteRace() {
  3.   for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
  4.     final PublishSubject<Integer> ps1 = PublishSubject.create();
  5.     final PublishSubject<Integer> ps2 = PublishSubject.create();
  6.     @SuppressWarnings("unchecked")
  7.     TestObserver<Integer> to = Observable.ambArray(ps1, ps2).test();
  8.     Runnable r1 = new Runnable() {
  9.       @Override
  10.       public void run() {
  11.         ps1.onComplete();
  12.       }
  13.     };
  14.     Runnable r2 = new Runnable() {
  15.       @Override
  16.       public void run() {
  17.         ps2.onComplete();
  18.       }
  19.     };
  20.     TestHelper.race(r1, r2);
  21.     to.assertResult();
  22.   }
  23. }

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

  1. TestObserver<Integer> to = Observable.ambArray(ps1, ps2).test();

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

  1. @Test
  2. public void onNextRace() {
  3.   for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
  4.     final PublishSubject<Integer> ps1 = PublishSubject.create();
  5.     final PublishSubject<Integer> ps2 = PublishSubject.create();
  6.     @SuppressWarnings("unchecked")
  7.     TestObserver<Integer> to = Observable.ambArray(ps1, ps2).test();
  8.     Runnable r1 = new Runnable() {
  9.       @Override
  10.       public void run() {
  11.         ps1.onNext(1);
  12.       }
  13.     };
  14.     Runnable r2 = new Runnable() {
  15.       @Override
  16.       public void run() {
  17.         ps2.onNext(1);
  18.       }
  19.     };
  20.     TestHelper.race(r1, r2);
  21.     to.assertSubscribed().assertNoErrors()
  22.     .assertNotComplete().assertValueCount(1);
  23.   }
  24. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void testAmbCancelsOthers() {
  4.   PublishSubject<Integer> source1 = PublishSubject.create();
  5.   PublishSubject<Integer> source2 = PublishSubject.create();
  6.   PublishSubject<Integer> source3 = PublishSubject.create();
  7.   TestObserver<Integer> to = new TestObserver<Integer>();
  8.   Observable.ambArray(source1, source2, source3).subscribe(to);
  9.   assertTrue("Source 1 doesn't have subscribers!", source1.hasObservers());
  10.   assertTrue("Source 2 doesn't have subscribers!", source2.hasObservers());
  11.   assertTrue("Source 3 doesn't have subscribers!", source3.hasObservers());
  12.   source1.onNext(1);
  13.   assertTrue("Source 1 doesn't have subscribers!", source1.hasObservers());
  14.   assertFalse("Source 2 still has subscribers!", source2.hasObservers());
  15.   assertFalse("Source 2 still has subscribers!", source3.hasObservers());
  16. }

代码示例来源：origin: akarnokd/akarnokd-misc

  1. @Test
  2.   @SuppressWarnings("unchecked")
  3.   public void test() throws Exception {
  4.     Observable<Object> obs1 = Observable
  5.         .<Object>just("obs 1 event")
  6.         .doOnSubscribe(s -> System.out.println("obs1 sub"))
  7.         .doOnDispose(() -> System.out.println("obs1 unsub"));
  9.     Observable<Object> obs2 = Observable
  10.         .<Object>just("obs 2 event")
  11.         .doOnSubscribe(s -> System.out.println("obs2 sub"))
  12.         .doOnDispose(() -> System.out.println("obs2 unsub"));
  14.     Observable
  15.     .ambArray(obs1, obs2)
  16.     .subscribe(System.out::println);
  18.     Thread.sleep(500);
  19.   }
  20. }

内容来源于网络，如有侵权，请联系作者删除！

Observable

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Observable类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com