Li Jun

java/util/concurrent/atomic/annotations.xml

@@ -0,0 +1,5 @@
+<root>
+  <item name='java.util.concurrent.atomic.AtomicInteger'>
+    <annotation name='android.support.annotation.NonNull' />
+  </item>
+</root>

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

@@ -16,6 +16,8 @@
 
 package cn.nextop.rxjava.share.practices;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
 
@@ -30,6 +32,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");
+//        // Wrong
+//        return observable.map( e -> { return new Tuple2<>(e.charAt(0) - "a".charAt(0) + 1, e); }  );
+
+//        // Bad, has side effect
+//        AtomicInteger a = new AtomicInteger(0);
+//        return observable.map( e -> { int index = a.addAndGet(1); return new Tuple2<>(index, e); }  );
+
+        return observable.map(e -> new Tuple2<Integer, String>(1, e)).scan( (pre, cur) -> new Tuple2<Integer, String>(pre.getV1() + 1, cur.getV2()));
     }
 }

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

@@ -19,9 +19,22 @@
 
 import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Observable;
+import io.reactivex.Observer;
+import io.reactivex.Scheduler;
 import io.reactivex.Single;
+import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subjects.PublishSubject;
+import javafx.collections.ObservableList;
 
+import java.util.Dictionary;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.security.auth.Subject;
 
 /**
  * @author Baoyi Chen
@@ -34,7 +47,28 @@ public class Practice2 {
      * 返回: Observable[("a", 2), ("b", 1), ("c", 2)]
      */
     public Observable<Tuple2<String, Integer>> wordCount1(Observable<String> words) {
-        throw new UnsupportedOperationException("implementation");
+//        // Passed, but not good
+//        Hashtable<String, Integer> info = new Hashtable<String, Integer>();
+//        words.map( e -> { Integer a = info.get(e); a = (a == null ? 1 : a+1); info.put(e, a); return e; } ).subscribe();
+//        System.out.println(info.toString());
+//        Observable< Tuple2<String, Integer> > ob = Observable.< Tuple2<String, Integer> >create( e -> {
+//            Enumeration<String> en = info.keys();
+//            while (en.hasMoreElements()) {
+//                String k = en.nextElement(); Integer v = info.get(k);
+//                e.onNext(new Tuple2<String, Integer>(k, v));
+//            }
+//            e.onComplete();
+//        });
+//
+//        return ob;
+
+        //AtomicInteger a = new AtomicInteger(0);
+        Observable< Tuple2<String, Integer> > ob = Observable.< Tuple2<String, Integer> >create( emitter -> {
+            words.groupBy( e -> e ).subscribe( e ->{ e.count().subscribe( count -> { emitter.onNext(new Tuple2<String, Integer>(e.getKey(), count.intValue())); } ); });
+            emitter.onComplete();
+        });
+
+        return ob;
     }
 
     /*
@@ -43,7 +77,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");
+        HashMap<String, Integer> info = new HashMap<String, Integer>();
+        words.groupBy( e -> e ).subscribe( e ->{ e.count().subscribe( count -> { info.put(e.getKey(), count.intValue()); } ); });
+
+        return Single.just(info);
     }
 
 }

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

@@ -28,7 +28,23 @@ public class Practice3 {
      * 根据iterate的结果求和
      */
     public Maybe<Integer> sum(Observable<Node> observable) {
-        throw new UnsupportedOperationException("implementation");
+//        // Trying codes
+//        return Maybe.create( emitter -> {
+//            iterate(observable).scan((x, y) -> {
+//                return x + y;
+//            }).subscribe( e -> {
+//                System.out.println(e.toString());
+//            } );
+//            observable.last().subscribe( e -> emitter.onSuccess(e) );
+//
+//            emitter.onComplete();
+//        });
+//        iterate(observable).scan((x, y) -> {
+//            return x + y;
+//        }).subscribe(e -> {
+//            System.out.println(e.toString());
+//        });
+        return iterate(observable).scan((x, y) -> x + y).lastElement();
     }
 
     /*
@@ -42,7 +58,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,8 @@
 
 
 import io.reactivex.Observable;
+import io.reactivex.Observer;
+import io.reactivex.schedulers.Schedulers;
 
 
 /**
@@ -44,7 +46,12 @@ public class Practice4 {
      *
      */
     public Observable<String> runInMultiThread(Observable<String> observable) {
-        throw new UnsupportedOperationException("implementation");
+        Observable ob = Observable.create(emitter -> {
+            observable.subscribe(e -> {
+                Observable.just(e).subscribeOn(Schedulers.io()).subscribe(a -> emitter.onNext(a));
+            });
+        });
+        return ob;
     }
 
 }

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

@@ -16,6 +16,7 @@
 
 package cn.nextop.rxjava.share.practices;
 
+import cn.nextop.rxjava.share.util.type.Tuple2;
 import io.reactivex.Maybe;
 import io.reactivex.Observable;
 import io.reactivex.Single;
@@ -35,7 +36,17 @@ public class Practice5 {
      * return: Single[3]
      */
     public Single<Long> count(Observable<String> source) {
-        throw new UnsupportedOperationException("implementation");
+//        Single<Long> single = Single.create(emitter -> {
+//            Observable<Tuple2<Long, String>> ob = Observable.create( emt -> {
+//                source.map(e -> { emt.onNext(new Tuple2<Long, String>(1L, e)); return e; });
+//            });
+//            ob.scan((last, current) -> {
+//                return new Tuple2<Long, String>(last.getV1() + current.getV1(), current.getV2());
+//            });
+//            ob.lastElement().subscribe(e -> emitter.onSuccess(e.getV1()));
+//        });
+//        return single;
+        return source.map( e -> new Tuple2<Long, String>(1L, e)).reduce(new Tuple2<Long, String>(0L, ""), (x, y) -> new Tuple2<Long, String>(x.getV1() + 1L, y.getV2())).map(e -> e.getV1());
     }
 
     /*
@@ -44,7 +55,10 @@ 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(e -> Observable.<String>create(emitter -> {
+                e.forEach(item -> emitter.onNext(item));
+                emitter.onComplete();
+            })).concatMap(e -> e);
     }
 
     /*
@@ -53,7 +67,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(e -> e).map(e -> e.getKey());
     }
 
     /*
@@ -62,7 +76,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.map(e -> { if (conditon.test(e)) { return Observable.just(e); } else { return Observable.<Integer>empty(); } }).concatMap(e -> e);
     }
 
     /*
@@ -71,7 +85,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.map(e -> new Tuple2<Integer, String>(0, e)).scan((pre, cur) -> new Tuple2<Integer, String>(pre.getV1() + 1, cur.getV2())).reduce((pre, cur) -> cur.getV1() == (Integer)index ? cur :
+                pre).map(e -> e.getV2());
     }
 
     /*
@@ -80,7 +95,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(1, count).concatMap(e -> source);
     }
 
     /*
@@ -89,7 +104,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 +113,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 +122,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.map(e -> Observable.just(e).delay(1, TimeUnit.SECONDS)).concatMap(e -> e);
     }
 
 }