Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Huang Wei #8

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Huang Wei #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
674a647
u
Mar 26, 2018
c3447da
u
Mar 26, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 9 additions & 8 deletions src/main/java/cn/nextop/rxjava/share/practices/Practice1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,13 @@
*/
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(p -> new Tuple2<Integer, String>(1, p))
.scan((p1, p2) -> new Tuple2<Integer, String>(1 + p1.getV1(), p2.getV2()));
}
}
45 changes: 25 additions & 20 deletions src/main/java/cn/nextop/rxjava/share/practices/Practice2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,34 +16,39 @@

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
*/
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(e -> e).flatMap(e -> e.count().toObservable().map(x -> Tuples.of(e.getKey(), x.intValue())));
}

/*
* 举例: 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::reducerFun);
}

private Map<String, Integer> reducerFun(Map<String, Integer> map, String s) {
Integer integer = map.get(s);
if (integer == null) integer = 0;
map.put(s, ++integer);
return map;
}
}
8 changes: 6 additions & 2 deletions src/main/java/cn/nextop/rxjava/share/practices/Practice3.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

/*
Expand All @@ -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 {
Expand Down
29 changes: 14 additions & 15 deletions src/main/java/cn/nextop/rxjava/share/practices/Practice4.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,7 @@


import io.reactivex.Observable;


import io.reactivex.schedulers.Schedulers;
/**
* @author Baoyi Chen
*/
Expand All @@ -30,21 +29,21 @@ public class Practice4 {
* 参数observable = Observable["a", "b", "c"]
* 参数observer在消费observable时，每个元素都在独立的线程
*
* thread 1 ---------------
* |-----------| ["a"] |
* | ---------------
* |
* ------------------------- ---------- |thread 2 ---------------
* |Observable["a","b","c"]|----|Observer|----|-----------| ["b"] |
* ------------------------- ---------- | ---------------
* |
* |thread 3 ---------------
* |-----------| ["c"] |
* ---------------
* thread 1 ---------------
* |-----------|Observer["a"]|
* | ---------------
* |
* ------------------------- |thread 2 ---------------
* |Observable["a","b","c"]|--------|-----------|Observer["b"]|
* ------------------------- | ---------------
* |
* |thread 3 ---------------
* |-----------|Observer["c"]|
* ---------------
*
*/
public Observable<String> runInMultiThread(Observable<String> observable) {
throw new UnsupportedOperationException("implementation");
public Observable<String> runInMultiThread(Observable<String> observable) {
return observable.flatMap(e -> Observable.just(e).subscribeOn(Schedulers.io()));
}

}
48 changes: 34 additions & 14 deletions src/main/java/cn/nextop/rxjava/share/practices/Practice5.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
*/
Expand All @@ -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);
}

/*
Expand All @@ -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));
}

/*
Expand All @@ -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());
}

/*
Expand All @@ -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();
});
}

/*
Expand All @@ -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();
}

/*
Expand All @@ -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);
}

/*
Expand All @@ -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);
}

/*
Expand All @@ -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));
}

}