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

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

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

Observable.map介绍

[英]Returns an Observable that applies a specified function to each item emitted by the source ObservableSource and emits the results of these function applications.

Scheduler: map does not operate by default on a particular Scheduler.
[中]返回一个Observable,它将指定的函数应用于源ObservableSource发出的每个项,并发出这些函数应用程序的结果。
Scheduler:map默认情况下不会在特定的计划程序上运行。

代码示例

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

  1. @Override
  2. public Observable<String> apply(Observable<Map<String, String>> o) {
  3. return o.map(new Function<Map<String, String>, String>() {
  4. @Override
  5. public String apply(Map<String, String> map) {
  6. return map.get("firstName");
  7. }
  8. });
  9. }

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

  1. @Override
  2. public Observable<Integer> apply(final Integer leftValue, Observable<Integer> rightValues) {
  3. return rightValues.map(new Function<Integer, Integer>() {
  4. @Override
  5. public Integer apply(Integer rightValue) throws Exception {
  6. return add.apply(leftValue, rightValue);
  7. }
  8. });
  9. }

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

  1. @Override
  2. public ObservableSource<Object> apply(Integer v) throws Exception {
  3. return Observable.range(1, 2).map(new Function<Integer, Object>() {
  4. @Override
  5. public Object apply(Integer w) throws Exception {
  6. throw new TestException();
  7. }
  8. });
  9. }
  10. })

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

  1. @Override
  2. public ObservableSource<Object> apply(Observable<Object> o) throws Exception {
  3. return o.map(Functions.identity());
  4. }
  5. });

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

  1. @Override
  2. public Object apply(Observable<Object> o) throws Exception {
  3. return o.map(Functions.identity());
  4. }
  5. }, false, 1, 1, 1);

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

  1. @Override
  2. public ObservableSource<Integer> apply(Integer v) throws Exception {
  3. return Observable.range(1, 2).map(new Function<Integer, Integer>() {
  4. @Override
  5. public Integer apply(Integer v) throws Exception {
  6. throw new TestException();
  7. }
  8. });
  9. }
  10. })

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

  1. @Override
  2. public ObservableSource<Integer> apply(Integer v) throws Exception {
  3. return Observable.range(1, 2).map(new Function<Integer, Integer>() {
  4. @Override
  5. public Integer apply(Integer w) throws Exception {
  6. throw new TestException();
  7. }
  8. });
  9. }
  10. }, true)

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

  1. private Observable<Integer> mergeNAsyncStreamsOfN(final int outerSize, final int innerSize) {
  2. Observable<Observable<Integer>> os = Observable.range(1, outerSize)
  3. .map(new Function<Integer, Observable<Integer>>() {
  4. @Override
  5. public Observable<Integer> apply(Integer i) {
  6. return Observable.range(1, innerSize).subscribeOn(Schedulers.computation());
  7. }
  8. });
  9. return Observable.merge(os);
  10. }

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

  1. @Override
  2. public Observable<String> apply(final GroupedObservable<Integer, Integer> group) {
  3. return group.subscribeOn(Schedulers.newThread()).map(new Function<Integer, String>() {
  4. @Override
  5. public String apply(Integer t1) {
  6. System.out.println("Received: " + t1 + " on group : " + group.getKey());
  7. return "first groups: " + t1;
  8. }
  9. });
  10. }

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

  1. @Test(expected = IllegalArgumentException.class)
  2. public void testMapWithIssue417() {
  3. Observable.just(1).observeOn(Schedulers.computation())
  4. .map(new Function<Integer, Integer>() {
  5. @Override
  6. public Integer apply(Integer arg0) {
  7. throw new IllegalArgumentException("any error");
  8. }
  9. }).blockingSingle();
  10. }

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

  1. private Observable<Integer> mergeNSyncStreamsOfN(final int outerSize, final int innerSize) {
  2. Observable<Observable<Integer>> os = Observable.range(1, outerSize)
  3. .map(new Function<Integer, Observable<Integer>>() {
  4. @Override
  5. public Observable<Integer> apply(Integer i) {
  6. return Observable.range(1, innerSize);
  7. }
  8. });
  9. return Observable.merge(os);
  10. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void mapReturnsNull() {
  3. just1.map(new Function<Integer, Object>() {
  4. @Override
  5. public Object apply(Integer v) {
  6. return null;
  7. }
  8. }).blockingSubscribe();
  9. }

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

  1. @Test
  2. public void testTake2() {
  3. Observable<Integer> o = Observable.just(1, 2, 3, 4, 5);
  4. Iterable<String> it = Arrays.asList("a", "b", "c", "d", "e");
  5. SquareStr squareStr = new SquareStr();
  6. o.map(squareStr).zipWith(it, concat2Strings).take(2).subscribe(printer);
  7. assertEquals(2, squareStr.counter.get());
  8. }

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

  1. /**
  2. * We expect IllegalStateException to pass thru map.
  3. */
  4. @Test(expected = IllegalStateException.class)
  5. public void testErrorPassesThruMap2() {
  6. Observable.error(new IllegalStateException()).map(new Function<Object, Object>() {
  7. @Override
  8. public Object apply(Object i) {
  9. return i;
  10. }
  11. }).blockingSingle();
  12. }

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

  1. @Override
  2. public Observable<String> apply(final String s) {
  3. return Observable.just(s)
  4. .mergeWith(Observable.interval(10, TimeUnit.MILLISECONDS)
  5. .map(new Function<Long, String>() {
  6. @Override
  7. public String apply(Long i) {
  8. return s + " " + i;
  9. }
  10. })).take(250);
  11. }
  12. })

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

  1. @Ignore("Not sure what this does")
  2. @Test
  3. public void addOnNextValueExceptionAdded() throws Exception {
  4. Observer<BadToString> observer = new BadToStringObserver();
  5. Observable.just(new BadToString(false))
  6. .map(new Function<BadToString, BadToString>() {
  7. @Override
  8. public BadToString apply(BadToString badToString) {
  9. throw new IllegalArgumentException("Failure while handling");
  10. }
  11. }).subscribe(observer);
  12. }

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

  1. @Ignore("Not sure what this does")
  2. @Test
  3. public void addOnNextValueExceptionNotAddedWithBadString() throws Exception {
  4. Observer<BadToString> observer = new BadToStringObserver();
  5. Observable.just(new BadToString(true))
  6. .map(new Function<BadToString, BadToString>() {
  7. @Override
  8. public BadToString apply(BadToString badToString) {
  9. throw new IllegalArgumentException("Failure while handling");
  10. }
  11. }).subscribe(observer);
  12. }

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

  1. @Test(expected = IllegalArgumentException.class)
  2. public void testTakeWithError() {
  3. Observable.fromIterable(Arrays.asList(1, 2, 3)).take(1)
  4. .map(new Function<Integer, Integer>() {
  5. @Override
  6. public Integer apply(Integer t1) {
  7. throw new IllegalArgumentException("some error");
  8. }
  9. }).blockingSingle();
  10. }

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

  1. @SuppressWarnings("unchecked")
  2. @Test
  3. public void source() {
  4. Observable<Integer> o = Observable.just(1);
  5. assertSame(o, ((HasUpstreamObservableSource<Integer>)o.map(Functions.<Integer>identity())).source());
  6. }
  7. }

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

  1. @Test
  2. public void fusedReject() {
  3. TestObserver<Integer> to = ObserverFusion.newTest(QueueFuseable.ANY | QueueFuseable.BOUNDARY);
  4. Observable.range(1, 5)
  5. .map(Functions.<Integer>identity())
  6. .subscribe(to);
  7. ObserverFusion.assertFusion(to, QueueFuseable.NONE)
  8. .assertResult(1, 2, 3, 4, 5);
  9. }

相关文章

Observable类方法