原文鏈接：
https://medium.com/@diolor/improving-ux-with-rxjava-4440a13b157f#.qdhu122d1</br>
翻譯： hanks
網絡永連接，服務器不出錯逞壁，培根沒肥肉
Search list animation by Daley P Maasz from Dribbble.
友好的用戶體驗通常是用戶很爽流济，但開發(fā)者很痛苦锐锣。 當用戶點擊一個按鈕后，因為后端沒有及時響應而卡住界面绳瘟，這回讓用戶很失望刺下。
現(xiàn)在讓我們創(chuàng)建一個更好的搜索功能（當用戶在EditText輸入文字時進行搜索）：
- 盡可能少的網絡請求
- 盡可能少的錯誤信息
Rx 的邏輯會十分簡單并且針對小的細節(jié)
讓我們從基本的邏輯開始:
讓用戶輸入文字時執(zhí)行網絡請求，當結果返回時進行顯示
RxTextView.textChanges(searchEditText)
.flatMap(Api::searchItems)
.subscribe(this::updateList, t->showError());
1. 減少網絡請求
有兩個問題：
- 1 . 輸入的文字每變化一個就執(zhí)行請求（很垃圾）稽荧，例如： 用戶很快的輸入 “a”（搜索”a“）橘茉，然后”b”（搜索“ab”），然后“c”（搜索”abc”）姨丈，然后又刪除”c”（搜索“ab”）畅卓， 輸入“e”（搜索”abe”）。這個過程執(zhí)行了5次請求蟋恬。
- 2： 假如網絡情況很差翁潘，多個線程同時進行，這時候就可能發(fā)生錯誤：如當用戶輸入“a”歼争，然后”ab”拜马，但是搜索“ab”的結果先返回了，之后返回了搜索“a的結果”沐绒，這時候輸入框的文字是”ab”俩莽，但結果卻是搜索“a”的
解決方案：
- 增加 throttling （節(jié)流）行為
debounce()的作用正是如此， 時間為 100-150ms 為佳乔遮，如果服務器返回需要300ms扮超，那么你可以在 500ms時更新UI界面
RxTextView.textChanges(searchEditText)
.debounce(150, MILLISECONDS)
.flatMap(Api::searchItems)
.subscribe(this::updateList, t->showError());
- 取消之前的請求:
使用switchMap來替代flatMap，switchMap停止之前發(fā)出的請求蹋肮， 假如在150ms的時候搜索”ab“出刷，在300ms的時候搜索了”abcd“，但是搜索”ab”的請求需要花費超過150ms坯辩，那么搜索“abcd”的請求開始的時候將會取消上一個請求馁龟，用戶只會獲取到最后的一次搜索的結果。
RxTextView.textChanges(searchEditText)
.debounce(150, MILLISECONDS)
.switchMap(Api::searchItems)
.subscribe(this::updateList, t->showError());
2. 沒有錯誤提示/沒有網絡錯誤提示
如果網絡請求失敗漆魔，將不再觀測文本的變化坷檩，（因為調用了OnError，整個事件流中斷）有送，這可以通過添加錯誤捕獲函數輕松搞定
RxTextView.textChanges(searchEditText)
.debounce(150, MILLISECONDS)
.switchMap(Api::searchItems)
.onErrorResumeNext(t-> empty())
.subscribe(this::updateList);
但是不要這樣做淌喻， 我們需要更好的處理方案僧家。假如 searchItems() api因為網絡連接而返回失敗的呢雀摘？ 或者是因為偶然的連接失敗呢？
我們需要 重試機制 :
RxTextView.textChanges(searchEditText)
.debounce(150, MILLISECONDS)
.switchMap(Api::searchItems)
.retryWhen(new RetryWithConnectivity())
.subscribe(this::updateList, t->showError());
如何進一步提升呢八拱？ 通過增加超時時間阵赠。 因為我們的 設計師 Lenzing 說 ”即使1s對用戶來說也很漫長”涯塔，于是我們有做了下面的事情：
RxTextView.textChanges(searchEditText)
.debounce(150, MILLISECONDS)
.switchMap(Api::searchItems)
.retryWhen(new RetryWithConnectivityIncremental(context, 5, 15, SECONDS))
.subscribe(this::updateList, t->showErrorToUser());
對比 RetryWithConnectivityIncremental 和 RetryWithConnectivity ， RetryWithConnectivityIncremental 更智能清蚀，它首先設置了 5 秒的超時時間匕荸， 如果這個時候連接失敗，如果用戶再次重試的話枷邪，超時時間會被設置的更長（如上面的15 秒）榛搔。
現(xiàn)在，使用RxJava 提示了用戶體驗东揣， 請求防抖動践惑， 取消上次請求獲取最新的結果， 智能的網絡超時重試嘶卧。這些可能用戶不會注意到尔觉，但是這是一個好的設計 ;)
標簽： AndroidRxjavaUX
demo的源碼在github
完整代碼：
public class MainActivity extends AppCompatActivity {
@Bind(R.id.et_keyword) EditText et_keyword;
@Bind(R.id.tv_result) TextView tv_result;
@Override protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
ButterKnife.bind(this);
RestAdapter retrofit = new RestAdapter.Builder().setEndpoint("https://suggest.taobao.com")
.setLogLevel(RestAdapter.LogLevel.FULL)
.setConverter(new GsonConverter(new Gson()))
.build();
final SearchService service = retrofit.create(SearchService.class);
RxTextView.textChanges(et_keyword)
// 上面的對 tv_result 的操作需要在主線程
.subscribeOn(AndroidSchedulers.mainThread())
.debounce(600, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
.filter(new Func1<CharSequence, Boolean>() {
@Override public Boolean call(CharSequence charSequence) {
// 清空搜索出來的結構
tv_result.setText("");
//當 EditText 中文字大于0的時候
return charSequence.length() > 0;
}
})
.switchMap(new Func1<CharSequence, Observable<Data>>() {
@Override public Observable<Data> call(CharSequence charSequence) {
// 搜索
return service.searchProdcut("utf-8", charSequence.toString());
}
})
// .retryWhen(new RetryWithConnectivityIncremental(MainActivity.this, 5, 15, TimeUnit.MILLISECONDS))
// 網絡操作在io線程
.subscribeOn(Schedulers.io())
//將 data 轉換成 ArrayList<ArrayList<String>>
.map(new Func1<Data, ArrayList<ArrayList<String>>>() {
@Override public ArrayList<ArrayList<String>> call(Data data) {
return data.result;
}
})
// 將 ArrayList<ArrayList<String>> 中每一項提取出來 ArrayList<String>
.flatMap(new Func1<ArrayList<ArrayList<String>>, Observable<ArrayList<String>>>() {
@Override public Observable<ArrayList<String>> call(ArrayList<ArrayList<String>> arrayLists) {
return Observable.from(arrayLists);
}
})
.filter(new Func1<ArrayList<String>, Boolean>() {
@Override public Boolean call(ArrayList<String> strings) {
return strings.size() >= 2;
}
})
.map(new Func1<ArrayList<String>, String>() {
@Override public String call(ArrayList<String> strings) {
return "[商品名稱:" + strings.get(0) + ", ID:" + strings.get(1) + "]\n";
}
})
// 發(fā)生錯誤后不要調用 onError，而是轉到 onErrorResumeNext
/*.onErrorResumeNext(new Func1<Throwable, Observable<? extends String>>() {
@Override public Observable<? extends String> call(Throwable throwable) {
return Observable.just("error result");
}
})*/
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override public void call(String charSequence) {
showpop(charSequence);
}
});
}
private void showpop(String result) {
tv_result.append(result);
}
interface SearchService {
@GET("/sug") Observable<Data> searchProdcut(@Query("code") String code, @Query("q") String keyword);
}
class Data {
public ArrayList<ArrayList<String>> result;
}
// https://suggest.taobao.com/sug?code=utf-8&q=%E6%89%8B%E6%9C%BA
}
BroadcastObservable.java
package com.hanks.rxsearch;
import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.net.ConnectivityManager;
import android.net.NetworkInfo;
import android.os.Looper;
import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.subscriptions.Subscriptions;
/**
* Created by hanks on 15-11-29.
*/
public class BroadcastObservable implements Observable.OnSubscribe<Boolean> {
private final Context context;
public BroadcastObservable(Context context) {
this.context = context;
}
public static Observable<Boolean> fromConnectivityManager(Context context) {
return Observable.create(new BroadcastObservable(context)).share();
}
private static Subscription unsubscribeInUiThread(final Action0 unsubscribe) {
return Subscriptions.create(new Action0() {
@Override public void call() {
if (Looper.getMainLooper() == Looper.myLooper()) {
unsubscribe.call();
} else {
final Scheduler.Worker inner = AndroidSchedulers.mainThread().createWorker();
inner.schedule(new Action0() {
@Override public void call() {
unsubscribe.call();
inner.unsubscribe();
}
});
}
}
});
}
@Override public void call(final Subscriber<? super Boolean> subscriber) {
final BroadcastReceiver receiver = new BroadcastReceiver() {
@Override public void onReceive(Context context, Intent intent) {
subscriber.onNext(isConnectedToInternet());
}
};
context.registerReceiver(receiver, new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));
subscriber.add(unsubscribeInUiThread(new Action0() {
@Override public void call() {
context.unregisterReceiver(receiver);
}
}));
}
private Boolean isConnectedToInternet() {
ConnectivityManager manager = (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);
NetworkInfo networkInfo = manager.getActiveNetworkInfo();
return networkInfo != null && networkInfo.isConnected();
}
}
RetryWithConnectivityIncremental.java
package com.hanks.rxsearch;
import android.content.Context;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import retrofit.RetrofitError;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
/**
* RetryWithConnectivityIncremental
* <p/>
* Created by hanks on 15-11-29.
*/
public class RetryWithConnectivityIncremental implements Func1<Observable<? extends Throwable>, Observable<?>> {
private final int startTimeOut;
private final int maxTimeOut;
private final TimeUnit timeUnit;
private int timeOut;
private Observable<Boolean> isConnected;
public RetryWithConnectivityIncremental(Context context, int startTimeOut, int maxTimeOut, TimeUnit timeUnit) {
this.startTimeOut = startTimeOut;
this.maxTimeOut = maxTimeOut;
this.timeOut = startTimeOut;
this.timeUnit = timeUnit;
isConnected = getConnectedObservable(context);
}
private Observable<Boolean> getConnectedObservable(Context context) {
return BroadcastObservable.fromConnectivityManager(context)
.distinctUntilChanged()
.filter(new Func1<Boolean, Boolean>() {
@Override public Boolean call(Boolean isConnected) {
return isConnected;
}
});
}
private Observable.Transformer<Boolean, Boolean> attachIncementalTimeOut() {
return new Observable.Transformer<Boolean, Boolean>() {
@Override public Observable<Boolean> call(Observable<Boolean> observable) {
return observable.timeout(timeOut, timeUnit).doOnError(new Action1<Throwable>() {
@Override public void call(Throwable throwable) {
if (throwable instanceof TimeoutException) {
timeOut = timeOut > maxTimeOut ? maxTimeOut : timeOut + startTimeOut;
}
}
});
}
};
}
@Override public Observable<?> call(final Observable<? extends Throwable> observable) {
return observable.flatMap(new Func1<Throwable, Observable<Boolean>>() {
@Override public Observable<Boolean> call(Throwable throwable) {
if (throwable instanceof RetrofitError && ((RetrofitError) throwable).getKind() == RetrofitError.Kind.NETWORK) {
return isConnected;
} else {
return Observable.error(throwable);
}
}
}).compose(attachIncementalTimeOut());
}
}
