Jiacheng Zheng

src/main/java/cn/nextop/rxjava/share/practices/Practice1.java

@@ -16,6 +16,7 @@
 
 package cn.nextop.rxjava.share.practices;
 
+import cn.nextop.rxjava.share.util.Lists;
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 
@@ -30,6 +31,13 @@ public class Practice1 {
      * 返回值 Observable[(1, "a"), (2, "b"), (3, "c")] 注意index从1开始
      */
     public Observable<Tuple2<Integer, String>> indexable(Observable<String> observable) {
-        throw new UnsupportedOperationException("implementation");
+        Observable.fromIterable(Lists.of("a", "b", "c")).subscribe(System.out::println);
+
+        return observable.map(s -> {
+            return new Tuple2<Integer, String>(1,s);
+        })
+        .scan((acc, t) -> {
+            return new Tuple2<Integer, String>(acc.getV1()+1, t.getV2());
+        });
     }
 }

src/main/java/cn/nextop/rxjava/share/practices/Practice2.java

@@ -17,10 +17,13 @@
 package cn.nextop.rxjava.share.practices;
 
 
+import cn.nextop.rxjava.share.util.Tuples;
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 import io.reactivex.Single;
+import io.reactivex.schedulers.Schedulers;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -34,7 +37,9 @@ public class Practice2 {
      * 返回: Observable[("a", 2), ("b", 1), ("c", 2)]
      */
     public Observable<Tuple2<String, Integer>> wordCount1(Observable<String> words) {
-        throw new UnsupportedOperationException("implementation");
+        return words.groupBy(s -> s).map(groups -> {
+            return groups.count().toObservable().map(c -> Tuples.of(groups.getKey(), c.intValue()));
+        }).flatMap(e->e);
     }
 
     /*
@@ -43,7 +48,10 @@ public Observable<Tuple2<String, Integer>> wordCount1(Observable<String> words)
      * 返回: Single[Map{a=2, b=1, c=2}]
      */
     public Single<Map<String, Integer>> wordCount2(Observable<String> words) {
-        throw new UnsupportedOperationException("implementation");
+        return this.wordCount1(words).reduce(new HashMap<String, Integer>(), (t1, t2) -> {
+            t1.put(t2.getV1(), t2.getV2());
+            return t1;
+        });
     }
 
 }

src/main/java/cn/nextop/rxjava/share/practices/Practice3.java

@@ -19,6 +19,9 @@
 import io.reactivex.Maybe;
 import io.reactivex.Observable;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * @author Baoyi Chen
  */
@@ -28,7 +31,7 @@ public class Practice3 {
      * 根据iterate的结果求和
      */
     public Maybe<Integer> sum(Observable<Node> observable) {
-        throw new UnsupportedOperationException("implementation");
+        return this.iterate(observable).reduce((x,y)-> x+y);
     }
 
     /*
@@ -42,7 +45,24 @@ public Maybe<Integer> sum(Observable<Node> observable) {
      * return Observable[4, 3, 6, 7, 5] 顺序无关
      */
     public Observable<Integer> iterate(Observable<Node> observable) {
-        throw new UnsupportedOperationException("implementation");
+        return Observable.create(e -> {
+            observable.subscribe(node -> {
+                List<Integer> arr = new ArrayList();
+                this._iterate(node, arr);
+                for(Integer num : arr) {
+                    e.onNext(num);
+                }
+            });
+            e.onComplete();
+        });
+    }
+
+    private void _iterate(Node node, List arr) {
+        if(node != null) {
+            _iterate(node.left, arr);
+            _iterate(node.right, arr);
+            arr.add(node.value);
+        }
     }
 
     public static class Node {

src/main/java/cn/nextop/rxjava/share/practices/Practice4.java

@@ -18,6 +18,7 @@
 
 
 import io.reactivex.Observable;
+import io.reactivex.schedulers.Schedulers;
 
 
 /**
@@ -44,7 +45,7 @@ public class Practice4 {
      *
      */
     public Observable<String> runInMultiThread(Observable<String> observable) {
-        throw new UnsupportedOperationException("implementation");
+        return observable.map(s -> Observable.just(s).subscribeOn(Schedulers.io())).flatMap(e->e);
     }
 
 }

src/main/java/cn/nextop/rxjava/share/practices/Practice5.java

@@ -16,10 +16,15 @@
 
 package cn.nextop.rxjava.share.practices;
 
+import cn.nextop.rxjava.share.util.Lists;
 import io.reactivex.Maybe;
 import io.reactivex.Observable;
+import io.reactivex.Scheduler;
 import io.reactivex.Single;
+import io.reactivex.schedulers.Schedulers;
 
+import java.lang.reflect.Array;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -35,7 +40,7 @@ public class Practice5 {
      * return: Single[3]
      */
     public Single<Long> count(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
+        return source.reduce((long)0, (acc, s) -> acc+1);
     }
 
     /*
@@ -44,7 +49,7 @@ public Single<Long> count(Observable<String> source) {
      * return: Observable["a", "b", "c","b", "c", "d"]
      */
     public Observable<String> convert(Observable<List<String>> source) {
-        throw new UnsupportedOperationException("implementation");
+        return source.map(l -> Observable.fromIterable(l)).flatMap(s->s);
     }
 
     /*
@@ -53,7 +58,7 @@ public Observable<String> convert(Observable<List<String>> source) {
      * return: Observable["a", "b", "c"]
      */
     public Observable<String> distinct(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
+        return new Practice2().wordCount1(source).map(t -> t.getV1());
     }
 
     /*
@@ -62,7 +67,7 @@ public Observable<String> distinct(Observable<String> source) {
      * return: Observable[3, 4]
      */
     public Observable<Integer> filter(Observable<Integer> source, Predicate<Integer> conditon) {
-        throw new UnsupportedOperationException("implementation");
+        return source.groupBy(n -> conditon.test(n)).buffer(2).map(arr -> arr.get(0).getKey() ? arr.get(0) : arr.get(1)).flatMap(s->s);
     }
 
     /*
@@ -71,7 +76,7 @@ public Observable<Integer> filter(Observable<Integer> source, Predicate<Integer>
      * return: Maybe[3]
      */
     public Maybe<String> elementAt(Observable<String> source, int index) {
-        throw new UnsupportedOperationException("implementation");
+        return Maybe.fromFuture(source.skip(index).take(1).toFuture());
     }
 
     /*
@@ -80,16 +85,23 @@ public Maybe<String> elementAt(Observable<String> source, int index) {
      * return: Observable["a", "b", "a", "b"]
      */
     public Observable<String> repeat(Observable<String> source, int count) {
-        throw new UnsupportedOperationException("implementation");
+        Observable head = Observable.empty();
+        for(int i=0;i<count;i++) {
+            head = Observable.concat(head, source);
+        }
+        return head;
     }
-
     /*
      * example:
      * param: Observable["a"], Observable["b"]
      * return: Observable["a", "b"]
      */
     public Observable<String> concat(List<Observable<String>> source) {
-        throw new UnsupportedOperationException("implementation");
+        List arr = new ArrayList();
+        for(Observable o : source) {
+            o.blockingSubscribe(s -> arr.add(s));
+        }
+        return Observable.fromIterable(arr);
     }
 
     /*
@@ -98,7 +110,7 @@ public Observable<String> concat(List<Observable<String>> source) {
      * return: Observable["a", "b"]
      */
     public Observable<String> merge(List<Observable<String>> source) {
-        throw new UnsupportedOperationException("implementation");
+        return Observable.fromIterable(source).flatMap(s->s);
     }
 
     /*
@@ -107,7 +119,14 @@ public Observable<String> merge(List<Observable<String>> source) {
      * return: Observable["a", "b", "c"], 每个元素都延迟1秒
      */
     public Observable<String> delayAll(Observable<String> source, long delay, TimeUnit unit) {
-        throw new UnsupportedOperationException("implementation");
+        return Observable.create(emiter -> {
+            source.forEach(s -> {
+                Observable.just(s).delay(delay, unit).blockingSubscribe(c -> {
+                    emiter.onNext(c);
+                });
+            });
+            emiter.onComplete();
+        });
     }
 
 }