Yunfei Liu

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.Tuples;
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 
@@ -30,6 +31,6 @@ 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");
+    	return observable.zipWith(Observable.range(1, Integer.MAX_VALUE), (v, n) -> Tuples.of(n, v));
     }
 }

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

@@ -17,12 +17,14 @@
 package cn.nextop.rxjava.share.practices;
 
 
+import java.util.HashMap;
+import java.util.Map;
+
+import cn.nextop.rxjava.share.util.Tuples;
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 import io.reactivex.Single;
 
-import java.util.Map;
-
 /**
  * @author Baoyi Chen
  */
@@ -34,7 +36,7 @@ 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(e -> e).flatMap(e -> e.count().toObservable().map(x -> Tuples.of(e.getKey(), x.intValue())));
     }
 
     /*
@@ -43,7 +45,9 @@ 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 wordCount1(words).reduce(new HashMap<String, Integer>(), (a, b) -> {a.put(b.getV1(), b.getV2());
+            return a;
+        });
     }
 
 }

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

@@ -28,7 +28,7 @@ public class Practice3 {
      * 根据iterate的结果求和
      */
     public Maybe<Integer> sum(Observable<Node> observable) {
-        throw new UnsupportedOperationException("implementation");
+    	return iterate(observable).reduce((x, y) -> x + y);
     }
 
     /*
@@ -42,7 +42,11 @@ 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.flatMap(n -> {
+            Observable<Integer> left = n.left == null ? Observable.empty() : iterate(Observable.just(n.left));
+            Observable<Integer> right = n.right == null ? Observable.empty() : iterate(Observable.just(n.right));
+            return Observable.merge(left, right, Observable.just(n.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.flatMap(e -> Observable.just(e).subscribeOn(Schedulers.newThread()));
     }
 
 }

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

@@ -16,14 +16,16 @@
 
 package cn.nextop.rxjava.share.practices;
 
-import io.reactivex.Maybe;
-import io.reactivex.Observable;
-import io.reactivex.Single;
-
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 
+import cn.nextop.rxjava.share.util.Tuples;
+import io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.ObservableEmitter;
+import io.reactivex.Single;
+
 /**
  * @author Baoyi Chen
  */
@@ -35,7 +37,8 @@ public class Practice5 {
      * return: Single[3]
      */
     public Single<Long> count(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
+//    	return source.count();
+    	return source.reduce(0L, (a, b) -> a + 1);
     }
 
     /*
@@ -44,7 +47,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.flatMap(e -> Observable.fromIterable(e));
     }
 
     /*
@@ -53,7 +56,8 @@ 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 source.distinct();
+    	return source.groupBy(e -> e).map(e -> e.getKey());
     }
 
     /*
@@ -62,7 +66,9 @@ 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.concatMap(e -> {
+            if (conditon.test(e)) return Observable.just(e); else return Observable.<Integer>empty();
+        });
     }
 
     /*
@@ -71,7 +77,8 @@ 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 source.elementAt(index);
+    	return source.zipWith(Observable.range(0, Integer.MAX_VALUE), (a, b) -> Tuples.of(a, b)).filter(x -> x.getV2() == index).map(e -> e.getV1()).firstElement();
     }
 
     /*
@@ -80,7 +87,8 @@ 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");
+//    	return source.repeat(count);
+    	return Observable.range(0, count).concatMap(e -> source);
     }
 
     /*
@@ -89,16 +97,27 @@ public Observable<String> repeat(Observable<String> source, int count) {
      * return: Observable["a", "b"]
      */
     public Observable<String> concat(List<Observable<String>> source) {
-        throw new UnsupportedOperationException("implementation");
+//    	return Observable.concat(source);
+    	return Observable.create(emitter -> { concat(source, emitter); });
+    }
+    private void concat(List<Observable<String>> source, ObservableEmitter<String> emitter) {
+        if (source.isEmpty()) {
+            emitter.onComplete();
+        } else {
+            source.get(0).subscribe(e -> {
+                emitter.onNext(e);
+            }, e -> emitter.onError(e), () -> {
+                concat(source.subList(1, source.size()), emitter);
+            });
+        }
     }
-
     /*
      * example:
      * param: Observable["a"], Observable["b"]
      * return: Observable["a", "b"]
      */
     public Observable<String> merge(List<Observable<String>> source) {
-        throw new UnsupportedOperationException("implementation");
+    	return Observable.fromIterable(source).flatMap(e -> e);
     }
 
     /*
@@ -107,7 +126,8 @@ 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 source.delay(delay, unit);
+    	return source.concatMap(e -> Observable.just(e).delay(delay, unit));
     }
 
 }