Yanming Qiu

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

@@ -29,7 +29,9 @@ 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");
+    public static Observable<Tuple2<Integer, String>> indexable(Observable<String> observable) {
+    	return observable.zipWith(Observable.range(1, Integer.MAX_VALUE), (a1,a2)->{return new Tuple2<>(a2,a1);});
     }
+
+
 }

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

@@ -17,10 +17,12 @@
 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 java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -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,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 words.reduce(new HashMap<String, Integer>(), (a, b) -> {
+             if (a.containsKey(b)) { a.put(b, a.get(b) + 1); } else { a.put(b, 1); }
+             return a;
+         });
     }
 
 }

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

@@ -23,13 +23,18 @@
  * @author Baoyi Chen
  */
 public class Practice3 {
-
+	  public static void main(String[] args) {
+			 Observable.just(4,3,5,6,7,8).reduce((x,y)->x+y).subscribe(e->System.out.println(e));;
+		}
     /*
      * 根据iterate的结果求和
      */
-    public Maybe<Integer> sum(Observable<Node> observable) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	public Maybe<Integer> sum(Observable<Node> observable) {
+		return this.iterate(observable).reduce((x, y) -> {
+			return x + y;
+		});
+
+	}
 
     /*
      * 举例:
@@ -42,7 +47,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");
+    	 Observable<Integer> ob = Observable.<Integer>create( emitter -> {
+	            observable.map( e -> {
+	                if (e.left != null) {
+	                    //left = iterate(Observable.just(e.left)).elementAt(0, Observable.just(0)).subscribe()
+	                    iterate(Observable.just(e.left)).subscribe(l -> emitter.onNext(l));
+	                }
+	                if (e.right != null) {
+	                    iterate(Observable.just(e.right)).subscribe(r -> emitter.onNext(r));
+	                }
+	                emitter.onNext(e.value);
+	                return e.value;
+	            }).subscribe();
+	            emitter.onComplete();
+	        });
+
+	        return ob;
     }
 
     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.concatMap(e->Observable.just(e).observeOn(Schedulers.newThread()));
     }
 
 }

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

@@ -16,14 +16,17 @@
 
 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.HashMap;
 import java.util.List;
+import java.util.Map;
 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 +38,7 @@ public class Practice5 {
      * return: Single[3]
      */
     public Single<Long> count(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
+        return source.toList().map(e->(long)e.size());
     }
 
     /*
@@ -44,25 +47,38 @@ 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");
+    	List<String> r = new ArrayList<>();
+    	source.subscribe(e->r.addAll(e));
+    	return Observable.fromIterable(r);
+
     }
 
     /*
      * example:
      * param: Observable["a", "a", "b", "b", "c"]
      * return: Observable["a", "b", "c"]
      */
-    public Observable<String> distinct(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
-    }
+	public Observable<String> distinct(Observable<String> source) {
+		Map<String, String> a = new HashMap<>();
+		return source.filter(e -> {
+			if (a.containsKey(e))
+				return false;
+			else {
+				a.put(e, e);
+				return true;
+			}
+		});
+	}
 
     /*
      * 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");
+    	return source.concatMap(e -> {
+            if (conditon.test(e)) return Observable.just(e); else return Observable.<Integer>empty();
+        });
     }
 
     /*
@@ -71,16 +87,17 @@ 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();
     }
 
+
     /*
      * 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");
+    	return Observable.range(0, count).concatMap(e -> source);
     }
 
     /*
@@ -89,7 +106,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(e -> e);
     }
 
     /*
@@ -98,7 +115,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(e -> e);
     }
 
     /*
@@ -107,7 +124,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(e -> Observable.just(e).delay(delay, unit));
     }
 
 }