Tianlian Qu

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;
 
@@ -24,12 +25,14 @@
  */
 public class Practice1 {
 
+	//
+
     /*
      * 举例如下:
      * 参数 Observable["a","b","c"]
      * 返回值 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), (i, s) -> Tuples.of(s, i));
     }
 }

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

@@ -17,11 +17,14 @@
 package cn.nextop.rxjava.share.practices;
 
 
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 import io.reactivex.Single;
-
-import java.util.Map;
+import io.reactivex.functions.BiFunction;
 
 /**
  * @author Baoyi Chen
@@ -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(key -> key).map(group -> {
+			return group.count().map(c -> new Tuple2<String, Integer>(group.getKey(), c.intValue())).toObservable();
+		}).flatMap(f -> f);
     }
 
     /*
@@ -43,7 +48,16 @@ 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 words.reduceWith(new Callable<Map<String, Integer>>() {
+    		@Override
+    		public Map<String, Integer> call() throws Exception {
+    			return new HashMap<String, Integer>();
+    		}
+		}, new BiFunction<Map<String, Integer>, String, Map<String, Integer>>() {
+			@Override
+			public Map<String, Integer> apply(Map<String, Integer> t1, String t2) throws Exception {
+				Integer c = t1.get(t2); if(c == null) c = 0; t1.put(t2, c++); return t1;
+			}
+		});
     }
-
 }

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

@@ -16,6 +16,9 @@
 
 package cn.nextop.rxjava.share.practices;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import io.reactivex.Maybe;
 import io.reactivex.Observable;
 
@@ -28,7 +31,7 @@ public class Practice3 {
      * 根据iterate的结果求和
      */
     public Maybe<Integer> sum(Observable<Node> observable) {
-        throw new UnsupportedOperationException("implementation");
+    	return this.iterate(observable).reduce((a,b) -> a + b);
     }
 
     /*
@@ -41,10 +44,16 @@ 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");
-    }
+	public Observable<Integer> iterate(Observable<Node> observable) {
+		return observable.flatMapIterable(n -> {
+			 List<Node> l = new ArrayList<>();
+			 fill(l, n); return l; }).map(n -> n.value);
+	}
 
+	private void fill(List<Node> r, Node n) {
+		if (n == null) return; r.add(n); fill(r, n.left); fill(r, n.right);
+	}
+
     public static class Node {
         public Node left;
         public Node right;

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.concatMap(s -> Observable.just(s).observeOn(Schedulers.newThread()));
     }
 
 }

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

@@ -16,14 +16,14 @@
 
 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 io.reactivex.Maybe;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+
 /**
  * @author Baoyi Chen
  */
@@ -35,7 +35,7 @@ public class Practice5 {
      * return: Single[3]
      */
     public Single<Long> count(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
+        return source.reduce(Long.valueOf("0"), (a, b) -> a + 1);
     }
 
     /*
@@ -44,7 +44,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.flatMapIterable(s -> s);
     }
 
     /*
@@ -53,7 +53,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 source.groupBy(s -> s).map(e -> e.getKey());
     }
 
     /*
@@ -62,7 +62,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.filter(conditon::test);
     }
 
     /*
@@ -71,7 +71,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 source.skip(index).take(1).firstElement();
     }
 
     /*
@@ -80,7 +80,7 @@ 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 Observable.range(0, count).concatMap(e -> source);
     }
 
     /*
@@ -89,7 +89,7 @@ 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.fromIterable(source).concatMap(s -> s);
     }
 
     /*
@@ -98,7 +98,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 +107,7 @@ 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.concatMap(s -> Observable.just(s).delay(delay, unit));
     }
 
 }