RxJava Mathematical and Aggregate operators

rxjava/pom.xml

@@ -25,6 +25,12 @@
             <version>2.1.3</version>
         </dependency>
 
+        <dependency>
+            <groupId>io.reactivex</groupId>
+            <artifactId>rxjava-math</artifactId>
+            <version>1.0.0</version>
+        </dependency>
+
         <dependency>
             <groupId>com.jayway.awaitility</groupId>
             <artifactId>awaitility</artifactId>

rxjava/src/test/java/com/baeldung/rxjava/operators/RxAggregateOperatorsTest.java

@@ -0,0 +1,210 @@
+package com.baeldung.rxjava.operators;
+
+import org.junit.Test;
+import rx.Observable;
+import rx.observers.TestSubscriber;
+
+import java.util.*;
+
+public class RxAggregateOperatorsTest {
+
+    @Test
+    public void givenTwoObservable_whenConcatenatingThem_thenSuccessfull() {
+        // given
+        List<Integer> listOne = Arrays.asList(1, 2, 3, 4);
+        Observable<Integer> observableOne = Observable.from(listOne);
+
+        List<Integer> listTwo = Arrays.asList(5, 6, 7, 8);
+        Observable<Integer> observableTwo = Observable.from(listTwo);
+
+        TestSubscriber<Integer> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<Integer> concatObservable = observableOne.concatWith(observableTwo);
+
+        concatObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(8);
+        subscriber.assertValues(1, 2, 3, 4, 5, 6, 7, 8);
+
+    }
+
+    @Test
+    public void givenObservable_whenCounting_thenObtainingNumberOfElements() {
+        // given
+        List<String> lettersList = Arrays.asList("A", "B", "C", "D", "E", "F", "G");
+
+        TestSubscriber<Integer> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<Integer> sourceObservable = Observable.from(lettersList)
+            .count();
+        sourceObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(7);
+    }
+
+    @Test
+    public void givenObservable_whenReducing_thenObtainingInvertedConcatenatedString() {
+        // given
+        List<String> list = Arrays.asList("A", "B", "C", "D", "E", "F", "G");
+
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<String> reduceObservable = Observable.from(list)
+            .reduce((letter1, letter2) -> letter2 + letter1);
+        reduceObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue("GFEDCBA");
+    }
+
+    @Test
+    public void givenObservable_whenCollecting_thenObtainingASet() {
+        // given
+        List<String> list = Arrays.asList("A", "B", "C", "B", "B", "A", "D");
+
+        TestSubscriber<HashSet> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<HashSet> reduceListObservable = Observable.from(list)
+            .collect(() -> new HashSet(), (set, item) -> set.add(item));
+        reduceListObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValues(new HashSet(list));
+    }
+
+    @Test
+    public void givenObservable_whenUsingToList_thenObtainedAList() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(1, 5);
+        TestSubscriber<List> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<List<Integer>> listObservable = sourceObservable.toList();
+        listObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(Arrays.asList(1, 2, 3, 4, 5));
+
+    }
+
+    @Test
+    public void givenObservable_whenUsingToSortedList_thenObtainedASortedList() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(10, 5);
+        TestSubscriber<List> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<List<Integer>> listObservable = sourceObservable.toSortedList();
+        listObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(Arrays.asList(10, 11, 12, 13, 14));
+    }
+
+    @Test
+    public void givenObservable_whenUsingToSortedListWithComparator_thenObtainedAnInverseSortedList() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(10, 5);
+        TestSubscriber<List> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<List<Integer>> listObservable = sourceObservable.toSortedList((int1, int2) -> int2 - int1);
+        listObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(Arrays.asList(14, 13, 12, 11, 10));
+
+    }
+
+    @Test
+    public void givenObservable_whenUsingToMap_thenObtainedAMap() {
+        // given
+        Observable<Book> bookObservable = Observable.just(new Book("The North Water", 2016), new Book("Origin", 2017), new Book("Sleeping Beauties", 2017));
+        TestSubscriber<Map> subscriber = TestSubscriber.create();
+
+        // when
+        Observable<Map<String, Integer>> mapObservable = bookObservable.toMap(Book::getTitle, Book::getYear, HashMap::new);
+
+        mapObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(new HashMap() {
+            {
+                put("The North Water", 2016);
+                put("Origin", 2017);
+                put("Sleeping Beauties", 2017);
+            }
+        });
+    }
+
+    @Test
+    public void givenObservable_whenUsingToMultiMap_thenObtainedAMultiMap() {
+        // given
+        Observable<Book> bookObservable = Observable.just(new Book("The North Water", 2016), new Book("Origin", 2017), new Book("Sleeping Beauties", 2017));
+        TestSubscriber<Map> subscriber = TestSubscriber.create();
+
+        // when
+        Observable multiMapObservable = bookObservable.toMultimap(Book::getYear, Book::getTitle, () -> new HashMap(), (key) -> new ArrayList());
+
+        multiMapObservable.subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(new HashMap() {
+            {
+                put(2016, Arrays.asList("The North Water"));
+                put(2017, Arrays.asList("Origin", "Sleeping Beauties"));
+            }
+        });
+
+    }
+
+    class Book {
+        private String title;
+        private Integer year;
+
+        public Book(String title, Integer year) {
+            this.title = title;
+            this.year = year;
+        }
+
+        public String getTitle() {
+            return title;
+        }
+
+        public Integer getYear() {
+            return year;
+        }
+    }
+}

rxjava/src/test/java/com/baeldung/rxjava/operators/RxMathematicalOperatorsTest.java

@@ -0,0 +1,139 @@
+package com.baeldung.rxjava.operators;
+
+import org.junit.Test;
+import rx.Observable;
+import rx.observables.MathObservable;
+import rx.observers.TestSubscriber;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+public class RxMathematicalOperatorsTest {
+
+    @Test
+    public void givenRangeNumericObservable_whenCalculatingAverage_ThenSuccessfull() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(1, 20);
+
+        TestSubscriber<Integer> subscriber = TestSubscriber.create();
+
+        // when
+        MathObservable.averageInteger(sourceObservable)
+            .subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(10);
+    }
+
+    @Test
+    public void givenRangeNumericObservable_whenCalculatingSum_ThenSuccessfull() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(1, 20);
+        TestSubscriber<Integer> subscriber = TestSubscriber.create();
+
+        // when
+        MathObservable.sumInteger(sourceObservable)
+            .subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(210);
+
+    }
+
+    @Test
+    public void givenRangeNumericObservable_whenCalculatingMax_ThenSuccessfullObtainingMaxValue() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(1, 20);
+        TestSubscriber<Integer> subscriber = TestSubscriber.create();
+
+        // when
+        MathObservable.max(sourceObservable)
+            .subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(20);
+    }
+
+    @Test
+    public void givenRangeNumericObservable_whenCalculatingMin_ThenSuccessfullObtainingMinValue() {
+        // given
+        Observable<Integer> sourceObservable = Observable.range(1, 20);
+        TestSubscriber<Integer> subscriber = TestSubscriber.create();
+
+        // when
+        MathObservable.min(sourceObservable)
+            .subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(1);
+    }
+
+    @Test
+    public void givenItemObservable_whenCalculatingMaxWithComparator_ThenSuccessfullObtainingMaxItem() {
+        // given
+        Item five = new Item(5);
+        List<Item> list = Arrays.asList(new Item(1), new Item(2), new Item(3), new Item(4), five);
+        Observable<Item> itemObservable = Observable.from(list);
+
+        TestSubscriber<Item> subscriber = TestSubscriber.create();
+
+        // when
+        MathObservable.from(itemObservable)
+            .max(Comparator.comparing(Item::getId))
+            .subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(five);
+
+    }
+
+    @Test
+    public void givenItemObservable_whenCalculatingMinWithComparator_ThenSuccessfullObtainingMinItem() {
+        // given
+        Item one = new Item(1);
+        List<Item> list = Arrays.asList(one, new Item(2), new Item(3), new Item(4), new Item(5));
+        TestSubscriber<Item> subscriber = TestSubscriber.create();
+        Observable<Item> itemObservable = Observable.from(list);
+
+        // when
+        MathObservable.from(itemObservable)
+            .min(Comparator.comparing(Item::getId))
+            .subscribe(subscriber);
+
+        // then
+        subscriber.assertCompleted();
+        subscriber.assertNoErrors();
+        subscriber.assertValueCount(1);
+        subscriber.assertValue(one);
+
+    }
+
+    class Item {
+        private Integer id;
+
+        public Item(Integer id) {
+            this.id = id;
+        }
+
+        public Integer getId() {
+            return id;
+        }
+
+    }
+}