Xinyang Pan

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

@@ -24,12 +24,12 @@
  */
 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");
-    }
+	/*
+	 * 举例如下:
+	 * 参数 Observable["a","b","c"]
+	 * 返回值 Observable[(1, "a"), (2, "b"), (3, "c")] 注意index从1开始
+	 */
+	public Observable<Tuple2<Integer, String>> indexable(Observable<String> observable) {
+		return observable.map(s -> new Tuple2<Integer, String>(1, s)).scan((t1, t2) -> new Tuple2<Integer, String>(t1.getV1() + 1, t2.getV2()));
+	}
 }

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

@@ -16,34 +16,45 @@
 
 package cn.nextop.rxjava.share.practices;
 
+import java.util.HashMap;
+import java.util.Map;
 
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 import io.reactivex.Single;
 
-import java.util.Map;
-
 /**
  * @author Baoyi Chen
  */
 public class Practice2 {
 
-    /*
-     * 举例:
-     * words = Observable["a", "a", "b", "c", "c"]
-     * 返回: Observable[("a", 2), ("b", 1), ("c", 2)]
-     */
-    public Observable<Tuple2<String, Integer>> wordCount1(Observable<String> words) {
-        throw new UnsupportedOperationException("implementation");
-    }
-
-    /*
-     * 举例:
-     * words = Observable["a", "a", "b", "c", "c"]
-     * 返回: Single[Map{a=2, b=1, c=2}]
-     */
-    public Single<Map<String, Integer>> wordCount2(Observable<String> words) {
-        throw new UnsupportedOperationException("implementation");
-    }
-
+	/*
+	 * 举例:
+	 * words = Observable["a", "a", "b", "c", "c"]
+	 * 返回: Observable[("a", 2), ("b", 1), ("c", 2)]
+	 */
+	public Observable<Tuple2<String, Integer>> wordCount1(Observable<String> words) {
+    	return words.groupBy(v -> v).map(gr -> {
+    		Observable<Tuple2<String, Integer>> map = gr.count().map(c -> new Tuple2<String, Integer>(gr.getKey(), c.intValue())).toObservable();
+			return map;
+    	}).flatMap(s -> s);
+	}
+
+	/*
+	 * 举例:
+	 * words = Observable["a", "a", "b", "c", "c"]
+	 * 返回: Single[Map{a=2, b=1, c=2}]
+	 */
+	public Single<Map<String, Integer>> wordCount2(Observable<String> words) {
+		return words.reduce(new HashMap<String, Integer>(), this::map);
+	}
+
+	private Map<String, Integer> map(Map<String, Integer> map, String s) {
+		Integer integer = map.get(s);
+		if (integer == null) {
+			integer = 0;
+		}
+		map.put(s, ++integer);
+		return map;
+	}
 }

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);
     }
 
     /*
@@ -42,9 +45,22 @@ 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.flatMapIterable(n -> {
+    		List<Node> l = new ArrayList<>();
+    		this.addLeafToList(l, n);
+    		return l;
+    	}).map(n -> n.value);
     }
 
+	private void addLeafToList(List<Node> l, Node n) {
+		if (n == null) {
+			return;
+		}
+		l.add(n);
+		this.addLeafToList(l, n.left);
+		this.addLeafToList(l, 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.concat(observable.map(s -> Observable.just(s).observeOn(Schedulers.io())));
     }
 
 }

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

@@ -16,98 +16,137 @@
 
 package cn.nextop.rxjava.share.practices;
 
-import io.reactivex.Maybe;
-import io.reactivex.Observable;
-import io.reactivex.Single;
-
+import java.util.ArrayList;
 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
  */
 public class Practice5 {
 
-    /*
-     * example:
-     * param: Observable["a","b","c"]
-     * return: Single[3]
-     */
-    public Single<Long> count(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
-    }
-
-    /*
-     * example:
-     * param: Observable[["a", "b", "c"], ["b", "c", "d"]]
-     * return: Observable["a", "b", "c","b", "c", "d"]
-     */
-    public Observable<String> convert(Observable<List<String>> source) {
-        throw new UnsupportedOperationException("implementation");
-    }
-
-    /*
-     * example:
-     * param: Observable["a", "a", "b", "b", "c"]
-     * return: Observable["a", "b", "c"]
-     */
-    public Observable<String> distinct(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable["a","b","c"]
+	 * return: Single[3]
+	 */
+	public Single<Long> count(Observable<String> source) {
+		return source.reduce(0L, (x, y) -> x + 1L);
+	}
 
-    /*
-     * example:
-     * param: Observable[1, 2, 3, 4, 5] , conditon = x > 2 and x < 5
-     * return: Observable[3, 4]
-     */
-    public Observable<Integer> filter(Observable<Integer> source, Predicate<Integer> conditon) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable[["a", "b", "c"], ["b", "c", "d"]]
+	 * return: Observable["a", "b", "c","b", "c", "d"]
+	 */
+	public Observable<String> convert(Observable<List<String>> source) {
+		return source.flatMapIterable(l -> l);
+	}
 
-    /*
-     * example:
-     * param: Observable[1, 2, 3, 4, 5] , index = 2
-     * return: Maybe[3]
-     */
-    public Maybe<String> elementAt(Observable<String> source, int index) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable["a", "a", "b", "b", "c"]
+	 * return: Observable["a", "b", "c"]
+	 */
+	public Observable<String> distinct(Observable<String> source) {
+		return source.groupBy(s -> s).map(g -> Observable.just(g.getKey())).flatMap(og -> og);
+	}
 
-    /*
-     * example:
-     * param: Observable["a", "b"] , count = 2
-     * return: Observable["a", "b", "a", "b"]
-     */
-    public Observable<String> repeat(Observable<String> source, int count) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable[1, 2, 3, 4, 5] , conditon = x > 2 and x < 5
+	 * return: Observable[3, 4]
+	 */
+	public Observable<Integer> filter(Observable<Integer> source, Predicate<Integer> conditon) {
+//		return source.map(s -> {
+//			if (conditon.test(s)) {
+//				return Observable.<Integer>just(s);
+//			} else {
+//				return Observable.<Integer>empty();
+//			}
+//		}).flatMap(o -> o);
+		return Observable.concat(source.map(s -> {
+			if (conditon.test(s)) {
+				return Observable.<Integer>just(s);
+			} else {
+				return Observable.<Integer>empty();
+			}
+		}));
+
+	}
 
-    /*
-     * example:
-     * param: Observable["a"], Observable["b"]
-     * return: Observable["a", "b"]
-     */
-    public Observable<String> concat(List<Observable<String>> source) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable[1, 2, 3, 4, 5] , index = 2
+	 * return: Maybe[3]
+	 */
+	public Maybe<String> elementAt(Observable<String> source, int index) {
+		return source.filter(new Index(index)).firstElement();
+	}
+
+	private class Index implements io.reactivex.functions.Predicate<String> {
+		private int i = 0;
+		private final int index;
+
+		private Index(int index) {
+			this.index = index;
+		}
+
+		@Override
+		public boolean test(String t) throws Exception {
+			if (i++ == index) {
+				return true;
+			}
+			return false;
+		}
+	}
+
+	/*
+	 * example:
+	 * param: Observable["a", "b"] , count = 2
+	 * return: Observable["a", "b", "a", "b"]
+	 */
+	public Observable<String> repeat(Observable<String> source, int count) {
+		List<Observable<String>> list = new ArrayList<>();
+		for (int i = 0; i < count; i++) {
+			list.add(source);
+		}
+		return Observable.concat(list);
+	}
 
-    /*
-     * example:
-     * param: Observable["a"], Observable["b"]
-     * return: Observable["a", "b"]
-     */
-    public Observable<String> merge(List<Observable<String>> source) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable["a"], Observable["b"]
+	 * return: Observable["a", "b"]
+	 */
+	public Observable<String> concat(List<Observable<String>> source) {
+		return Observable.fromIterable(source).concatMapDelayError(s -> s);
+	}
+
+	/*
+	 * example:
+	 * param: Observable["a"], Observable["b"]
+	 * return: Observable["a", "b"]
+	 */
+	public Observable<String> merge(List<Observable<String>> source) {
+		return Observable.fromIterable(source).flatMap(s -> s);
+	}
 
-    /*
-     * example:
-     * param: Observable["a", "b", "c"], 1, SECONDS
-     * return: Observable["a", "b", "c"], 每个元素都延迟1秒
-     */
-    public Observable<String> delayAll(Observable<String> source, long delay, TimeUnit unit) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	/*
+	 * example:
+	 * param: Observable["a", "b", "c"], 1, SECONDS
+	 * return: Observable["a", "b", "c"], 每个元素都延迟1秒
+	 */
+	public Observable<String> delayAll(Observable<String> source, long delay, TimeUnit unit) {
+		return source.map(t -> {
+			unit.sleep(delay);
+			return t;
+		});
+	}
 
 }