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

Observable.switchMap介绍

[英]Returns a new ObservableSource by applying a function that you supply to each item emitted by the source ObservableSource that returns an ObservableSource, and then emitting the items emitted by the most recently emitted of these ObservableSources.

The resulting ObservableSource completes if both the upstream ObservableSource and the last inner ObservableSource, if any, complete. If the upstream ObservableSource signals an onError, the inner ObservableSource is disposed and the error delivered in-sequence.

Scheduler: switchMap does not operate by default on a particular Scheduler.
[中]通过向源ObservableSource（返回ObservableSource的源ObservableSource）发出的每个项应用一个函数，然后发出这些ObservableSource中最近发出的项，返回一个新的ObservableSource。
如果上游可观测资源和最后一个内部可观测资源（如果有）都已完成，则生成的可观测资源将完成。如果上游可观测资源发出onError信号，内部可观测资源将被处理，错误将按顺序传递。
调度器：switchMap默认情况下不会在特定的调度器上运行。

代码示例

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

@Override
  public ObservableSource<Object> apply(Observable<Throwable> errors) throws Exception {
    return errors.switchMap(new Function<Throwable, ObservableSource<Object>>() {
      @Override
      public ObservableSource<Object> apply(Throwable ignore) throws Exception {
        return subject;
      }
    });
  }
})

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

@Override
  public ObservableSource<Object> apply(Observable<Object> completions) throws Exception {
    return completions.switchMap(new Function<Object, ObservableSource<Object>>() {
      @Override
      public ObservableSource<Object> apply(Object ignore) throws Exception {
        return subject;
      }
    });
  }
})

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

@Test(expected = NullPointerException.class)
public void switchMapNull() {
  just1.switchMap(null);
}

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

/**
 * Returns a new ObservableSource by applying a function that you supply to each item emitted by the source
 * ObservableSource that returns an ObservableSource, and then emitting the items emitted by the most recently emitted
 * of these ObservableSources.
 * <p>
 * The resulting ObservableSource completes if both the upstream ObservableSource and the last inner ObservableSource, if any, complete.
 * If the upstream ObservableSource signals an onError, the inner ObservableSource is disposed and the error delivered in-sequence.
 * <p>
 * <img width="640" height="350" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchMap.png" alt="">
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code switchMap} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 *
 * @param <R> the element type of the inner ObservableSources and the output
 * @param mapper
 *            a function that, when applied to an item emitted by the source ObservableSource, returns an
 *            ObservableSource
 * @return an Observable that emits the items emitted by the ObservableSource returned from applying {@code func} to the most recently emitted item emitted by the source ObservableSource
 * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> switchMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
  return switchMap(mapper, bufferSize());
}

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

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

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

@Test
public void switchMapJustSource() {
  Observable.just(0)
  .switchMap(new Function<Object, ObservableSource<Integer>>() {
    @Override
    public ObservableSource<Integer> apply(Object v) throws Exception {
      return Observable.just(1);
    }
  }, 16)
  .test()
  .assertResult(1);
}

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

@Test
public void mapperThrows() {
  Observable.just(1).hide()
  .switchMap(new Function<Integer, ObservableSource<Object>>() {
    @Override
    public ObservableSource<Object> apply(Integer v) throws Exception {
      throw new TestException();
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void asyncFused() {
  Observable.just(1).hide()
  .switchMap(Functions.justFunction(
      Observable.range(1, 5)
      .observeOn(ImmediateThinScheduler.INSTANCE)
  ))
  .test()
  .assertResult(1, 2, 3, 4, 5);
}

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

@Test
public void justInner() {
  Observable.range(1, 5)
  .switchMap(Functions.justFunction(Observable.just(1)))
  .test()
  .assertResult(1, 1, 1, 1, 1);
}

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

@Test
public void switchMapErrorEmptySource() {
  assertSame(Observable.empty(), Observable.<Object>empty()
      .switchMap(new Function<Object, ObservableSource<Integer>>() {
        @Override
        public ObservableSource<Integer> apply(Object v) throws Exception {
          return Observable.just(1);
        }
      }, 16));
}

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

@Test
public void innerCompletesReentrant() {
  final PublishSubject<Integer> ps = PublishSubject.create();
  TestObserver<Integer> to = new TestObserver<Integer>() {
    @Override
    public void onNext(Integer t) {
      super.onNext(t);
      ps.onComplete();
    }
  };
  Observable.just(1).hide()
  .switchMap(Functions.justFunction(ps))
  .subscribe(to);
  ps.onNext(1);
  to.assertResult(1);
}

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

@Test
public void innerErrorsReentrant() {
  final PublishSubject<Integer> ps = PublishSubject.create();
  TestObserver<Integer> to = new TestObserver<Integer>() {
    @Override
    public void onNext(Integer t) {
      super.onNext(t);
      ps.onError(new TestException());
    }
  };
  Observable.just(1).hide()
  .switchMap(Functions.justFunction(ps))
  .subscribe(to);
  ps.onNext(1);
  to.assertFailure(TestException.class, 1);
}

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

@Test
public void asyncFusedRejecting() {
  Observable.just(1).hide()
  .switchMap(Functions.justFunction(
      TestHelper.rejectObservableFusion()
  ))
  .test()
  .assertEmpty();
}

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

@Test
  public void fusedBoundary() {
    String thread = Thread.currentThread().getName();

    Observable.range(1, 10000)
    .switchMap(new Function<Integer, ObservableSource<? extends Object>>() {
      @Override
      public ObservableSource<? extends Object> apply(Integer v)
          throws Exception {
        return Observable.just(2).hide()
        .observeOn(Schedulers.single())
        .map(new Function<Integer, Object>() {
          @Override
          public Object apply(Integer w) throws Exception {
            return Thread.currentThread().getName();
          }
        });
      }
    })
    .test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertNever(thread)
    .assertNoErrors()
    .assertComplete();
  }
}

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

@Test
public void emptyInner() {
  Observable.range(1, 5)
  .switchMap(Functions.justFunction(Observable.empty()))
  .test()
  .assertResult();
}

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

@Test
public void switchMapInnerCancelled() {
  PublishSubject<Integer> ps = PublishSubject.create();
  TestObserver<Integer> to = Observable.just(1)
      .switchMap(Functions.justFunction(ps))
      .test();
  assertTrue(ps.hasObservers());
  to.cancel();
  assertFalse(ps.hasObservers());
}

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

@Test
public void badInnerSource() {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    Observable.just(1).hide()
    .switchMap(Functions.justFunction(new Observable<Integer>() {
      @Override
      protected void subscribeActual(Observer<? super Integer> observer) {
        observer.onSubscribe(Disposables.empty());
        observer.onError(new TestException());
        observer.onComplete();
        observer.onError(new TestException());
        observer.onComplete();
      }
    }))
    .test()
    .assertFailure(TestException.class);
    TestHelper.assertUndeliverable(errors, 0, TestException.class);
  } finally {
    RxJavaPlugins.reset();
  }
}

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

@Test
public void syncFusedSingle() {
  Observable.range(1, 5).hide()
  .switchMap(Functions.justFunction(
      Single.just(1).toObservable()
  ))
  .test()
  .assertResult(1, 1, 1, 1, 1);
}

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

@Test
public void syncFusedCompletable() {
  Observable.range(1, 5).hide()
  .switchMap(Functions.justFunction(
      Completable.complete().toObservable()
  ))
  .test()
  .assertResult();
}

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

@Test
public void syncFusedMaybe() {
  Observable.range(1, 5).hide()
  .switchMap(Functions.justFunction(
      Maybe.just(1).toObservable()
  ))
  .test()
  .assertResult(1, 1, 1, 1, 1);
}