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

Observable.flatMapMaybe介绍

[英]Maps each element of the upstream Observable into MaybeSources, subscribes to all of them and merges their onSuccess values, in no particular order, into a single Observable sequence.

Scheduler: flatMapMaybe does not operate by default on a particular Scheduler.
[中]将上游可观测的每个元素映射到可能资源中，订阅所有这些元素，并将它们的onSuccess值（无特定顺序）合并到单个可观测序列中。
调度器：FlatMap默认情况下可能不会在特定的调度器上运行。

代码示例

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

/**
 * Maps each element of the upstream Observable into MaybeSources, subscribes to all of them
 * and merges their onSuccess values, in no particular order, into a single Observable sequence.
 * <p>
 * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/flatMapMaybe.png" alt="">
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code flatMapMaybe} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * @param <R> the result value type
 * @param mapper the function that received each source value and transforms them into MaybeSources.
 * @return the new Observable instance
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMapMaybe(Function<? super T, ? extends MaybeSource<? extends R>> mapper) {
  return flatMapMaybe(mapper, false);
}

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

@Override
  public ObservableSource<Integer> apply(Observable<Object> f) throws Exception {
    return f.flatMapMaybe(Functions.justFunction(Maybe.just(2)));
  }
});

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

@Test
  public void disposeInner() {
    final TestObserver<Object> to = new TestObserver<Object>();

    Observable.just(1).flatMapMaybe(new Function<Integer, MaybeSource<Object>>() {
      @Override
      public MaybeSource<Object> apply(Integer v) throws Exception {
        return new Maybe<Object>() {
          @Override
          protected void subscribeActual(MaybeObserver<? super Object> observer) {
            observer.onSubscribe(Disposables.empty());

            assertFalse(((Disposable)observer).isDisposed());

            to.dispose();

            assertTrue(((Disposable)observer).isDisposed());
          }
        };
      }
    })
    .subscribe(to);

    to
    .assertEmpty();
  }
}

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

@Test
public void observerMaybe() throws Exception {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    TestObserver<Integer> to = Observable.just(1)
    .subscribeOn(Schedulers.io())
    .flatMapMaybe(new Function<Integer, Maybe<Integer>>() {
      @Override
      public Maybe<Integer> apply(Integer v) throws Exception {
        sleep();
        return Maybe.<Integer>error(new TestException());
      }
    })
    .test();
    cb.await();
    beforeCancelSleep(to);
    to.cancel();
    Thread.sleep(SLEEP_AFTER_CANCEL);
    to.assertEmpty();
    assertTrue(errors.toString(), errors.isEmpty());
  } finally {
    RxJavaPlugins.reset();
  }
}

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

@Test
public void normal() {
  Observable.range(1, 10)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.just(v);
    }
  })
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

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

@Test
public void normalEmpty() {
  Observable.range(1, 10)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.empty();
    }
  })
  .test()
  .assertResult();
}

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

@Test
public void normalDelayError() {
  Observable.range(1, 10)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.just(v);
    }
  }, true)
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

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

@Test
public void middleError() {
  Observable.fromArray(new String[]{"1", "a", "2"}).flatMapMaybe(new Function<String, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(final String s) throws NumberFormatException {
      //return Single.just(Integer.valueOf(s)); //This works
      return Maybe.fromCallable(new Callable<Integer>() {
        @Override
        public Integer call() throws NumberFormatException {
          return Integer.valueOf(s);
        }
      });
    }
  })
  .test()
  .assertFailure(NumberFormatException.class, 1);
}

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

@Test
public void take() {
  Observable.range(1, 10)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.just(v);
    }
  })
  .take(2)
  .test()
  .assertResult(1, 2);
}

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

@Test
public void innerSuccessCompletesAfterMain() {
  PublishSubject<Integer> ps = PublishSubject.create();
  TestObserver<Integer> to = Observable.just(1).flatMapMaybe(Functions.justFunction(ps.singleElement()))
  .test();
  ps.onNext(2);
  ps.onComplete();
  to
  .assertResult(2);
}

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

@Test
public void emissionQueueTrigger() {
  final PublishSubject<Integer> ps1 = PublishSubject.create();
  final PublishSubject<Integer> ps2 = PublishSubject.create();
  TestObserver<Integer> to = new TestObserver<Integer>() {
    @Override
    public void onNext(Integer t) {
      super.onNext(t);
      if (t == 1) {
        ps2.onNext(2);
        ps2.onComplete();
      }
    }
  };
  Observable.just(ps1, ps2)
      .flatMapMaybe(new Function<PublishSubject<Integer>, MaybeSource<Integer>>() {
        @Override
        public MaybeSource<Integer> apply(PublishSubject<Integer> v) throws Exception {
          return v.singleElement();
        }
      })
  .subscribe(to);
  ps1.onNext(1);
  ps1.onComplete();
  to.assertResult(1, 2);
}

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

@Test
public void asyncFlattenNone() {
  Observable.range(1, 1000)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.<Integer>empty().subscribeOn(Schedulers.computation());
    }
  })
  .take(500)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult();
}

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

@Test
public void normalAsync() {
  Observable.range(1, 10)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.just(v).subscribeOn(Schedulers.computation());
    }
  })
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  .assertNoErrors()
  .assertComplete();
}

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

@Test
public void emissionQueueTrigger2() {
  final PublishSubject<Integer> ps1 = PublishSubject.create();
  final PublishSubject<Integer> ps2 = PublishSubject.create();
  final PublishSubject<Integer> ps3 = PublishSubject.create();
  TestObserver<Integer> to = new TestObserver<Integer>() {
    @Override
    public void onNext(Integer t) {
      super.onNext(t);
      if (t == 1) {
        ps2.onNext(2);
        ps2.onComplete();
      }
    }
  };
  Observable.just(ps1, ps2, ps3)
      .flatMapMaybe(new Function<PublishSubject<Integer>, MaybeSource<Integer>>() {
        @Override
        public MaybeSource<Integer> apply(PublishSubject<Integer> v) throws Exception {
          return v.singleElement();
        }
      })
  .subscribe(to);
  ps1.onNext(1);
  ps1.onComplete();
  ps3.onComplete();
  to.assertResult(1, 2);
}

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

@Test
public void badInnerSource() {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    Observable.just(1)
    .flatMapMaybe(Functions.justFunction(new Maybe<Integer>() {
      @Override
      protected void subscribeActual(MaybeObserver<? super Integer> observer) {
        observer.onSubscribe(Disposables.empty());
        observer.onError(new TestException("First"));
        observer.onError(new TestException("Second"));
      }
    }))
    .test()
    .assertFailureAndMessage(TestException.class, "First");
    TestHelper.assertUndeliverable(errors, 0, TestException.class, "Second");
  } finally {
    RxJavaPlugins.reset();
  }
}

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

@Test
public void completeError() {
  final PublishSubject<Integer> ps = PublishSubject.create();
  TestObserver<Integer> to = Observable.range(1, 2)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      if (v == 2) {
        return ps.singleElement();
      }
      return Maybe.error(new TestException());
    }
  }, true)
  .test();
  ps.onComplete();
  to
  .assertFailure(TestException.class);
}

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

@Test
public void takeAsync() {
  Observable.range(1, 10)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.just(v).subscribeOn(Schedulers.computation());
    }
  })
  .take(2)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueCount(2)
  .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  .assertNoErrors()
  .assertComplete();
}

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

@Test
public void normalDelayErrorAll() {
  TestObserver<Integer> to = Observable.range(1, 10).concatWith(Observable.<Integer>error(new TestException()))
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.error(new TestException());
    }
  }, true)
  .test()
  .assertFailure(CompositeException.class);
  List<Throwable> errors = TestHelper.compositeList(to.errors().get(0));
  for (int i = 0; i < 11; i++) {
    TestHelper.assertError(errors, i, TestException.class);
  }
}

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

@Test
public void successError() {
  final PublishSubject<Integer> ps = PublishSubject.create();
  TestObserver<Integer> to = Observable.range(1, 2)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      if (v == 2) {
        return ps.singleElement();
      }
      return Maybe.error(new TestException());
    }
  }, true)
  .test();
  ps.onNext(1);
  ps.onComplete();
  to
  .assertFailure(TestException.class, 1);
}

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

@Test
public void asyncFlatten() {
  Observable.range(1, 1000)
  .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
    @Override
    public MaybeSource<Integer> apply(Integer v) throws Exception {
      return Maybe.just(1).subscribeOn(Schedulers.computation());
    }
  })
  .take(500)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueCount(500)
  .assertNoErrors()
  .assertComplete();
}