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

Flowable.reduce介绍

[英]Returns a Maybe that applies a specified accumulator function to the first item emitted by a source Publisher, then feeds the result of that function along with the second item emitted by the source Publisher into the same function, and so on until all items have been emitted by the finite source Publisher, and emits the final result from the final call to your function as its sole item.

If the source is empty, a NoSuchElementException is signaled.

This technique, which is called "reduce" here, is sometimes called "aggregate," "fold," "accumulate," "compress," or "inject" in other programming contexts. Groovy, for instance, has an inject method that does a similar operation on lists.

Note that this operator requires the upstream to signal onComplete for the accumulator object to be emitted. Sources that are infinite and never complete will never emit anything through this operator and an infinite source may lead to a fatal OutOfMemoryError. Backpressure: The operator honors backpressure of its downstream consumer and consumes the upstream source in unbounded mode. Scheduler: reduce does not operate by default on a particular Scheduler.
[中]

代码示例

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

@Override
  public MaybeSource<Integer> apply(Flowable<Integer> f) throws Exception {
    return f.reduce(sum);
  }
});

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

public void libraryFunctionActingOnMovieObservables(Flowable<Movie> obs) {
  obs.reduce(new BiFunction<Movie, Movie, Movie>() {
    @Override
    public Movie apply(Movie t1, Movie t2) {
      return t2;
    }
  });
}

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

@Override
  public Object apply(Flowable<Integer> f) throws Exception {
    return f.reduce(sum);
  }
}, false, 1, 1, 1);

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

@Override
  public SingleSource<Integer> apply(Flowable<Integer> f)
      throws Exception {
    return f.reduce(0, new BiFunction<Integer, Integer, Integer>() {
      @Override
      public Integer apply(Integer a, Integer b) throws Exception {
        return a;
      }
    });
  }
});

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

@Override
  public Object apply(Flowable<Integer> f) throws Exception {
    return f.reduce(sum).toFlowable();
  }
}, false, 1, 1, 1);

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

@Test(expected = NullPointerException.class)
public void reduceSeedNull() {
  just1.reduce(null, new BiFunction<Object, Integer, Object>() {
    @Override
    public Object apply(Object a, Integer b) {
      return 1;
    }
  });
}

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

@Test(expected = NullPointerException.class)
public void reduceSeedFunctionReturnsNull() {
  just1.reduce(1, new BiFunction<Integer, Integer, Integer>() {
    @Override
    public Integer apply(Integer a, Integer b) {
      return null;
    }
  }).blockingGet();
}

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

@Override
public Publisher<Integer> createPublisher(final long elements) {
  return
      Flowable.range(1, 1000).reduce(new BiFunction<Integer, Integer, Integer>() {
        @Override
        public Integer apply(Integer a, Integer b) throws Exception {
          return a + b;
        }
      }).toFlowable()
    ;
}

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

@Test
public void testBackpressureWithNoInitialValue() throws InterruptedException {
  Flowable<Integer> source = Flowable.just(1, 2, 3, 4, 5, 6);
  Maybe<Integer> reduced = source.reduce(sum);
  Integer r = reduced.blockingGet();
  assertEquals(21, r.intValue());
}

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

@Test
public void testBackpressureWithNoInitialValueObservable() throws InterruptedException {
  Flowable<Integer> source = Flowable.just(1, 2, 3, 4, 5, 6);
  Flowable<Integer> reduced = source.reduce(sum).toFlowable();
  Integer r = reduced.blockingFirst();
  assertEquals(21, r.intValue());
}

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

@Test
public void testBackpressureWithInitialValueFlowable() throws InterruptedException {
  Flowable<Integer> source = Flowable.just(1, 2, 3, 4, 5, 6);
  Flowable<Integer> reduced = source.reduce(0, sum).toFlowable();
  Integer r = reduced.blockingFirst();
  assertEquals(21, r.intValue());
}

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

@Test(expected = NullPointerException.class)
public void reduceFunctionReturnsNull() {
  Flowable.just(1, 1).reduce(new BiFunction<Integer, Integer, Integer>() {
    @Override
    public Integer apply(Integer a, Integer b) {
      return null;
    }
  }).toFlowable().blockingSubscribe();
}

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

@Test
public void empty() {
  Flowable.<Integer>empty()
  .reduce(sum)
  .test()
  .assertResult();
}

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

@Test
public void reducerThrows() {
  Flowable.just(1, 2)
  .reduce(new BiFunction<Integer, Integer, Integer>() {
    @Override
    public Integer apply(Integer a, Integer b) throws Exception {
      throw new TestException();
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void source() {
  Flowable<Integer> source = Flowable.just(1);
  assertSame(source, (((HasUpstreamPublisher<?>)source.reduce(sum))).source());
}

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

@Test
public void reduceIntsFlowable() {
  Flowable<Integer> f = Flowable.just(1, 2, 3);
  int value = f.reduce(new BiFunction<Integer, Integer, Integer>() {
    @Override
    public Integer apply(Integer t1, Integer t2) {
      return t1 + t2;
    }
  }).toFlowable().blockingSingle();
  assertEquals(6, value);
}

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

@Test
public void reduceInts() {
  Flowable<Integer> f = Flowable.just(1, 2, 3);
  int value = f.reduce(new BiFunction<Integer, Integer, Integer>() {
    @Override
    public Integer apply(Integer t1, Integer t2) {
      return t1 + t2;
    }
  }).toFlowable().blockingSingle();
  assertEquals(6, value);
}

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

@Test
public void error() {
  Flowable.<Integer>error(new TestException())
  .reduce(sum)
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void emptyFlowable() {
  Flowable.<Integer>empty()
  .reduce(sum)
  .toFlowable()
  .test()
  .assertResult();
}

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

@Test
public void errorFlowable() {
  Flowable.<Integer>error(new TestException())
  .reduce(sum)
  .toFlowable()
  .test()
  .assertFailure(TestException.class);
}