译自: May 2017 Meetup: Advanced RxJava and Conductor
Slides

前言

我猜在你学习 RxJava + Retrofit 的历程中，肯定见过这种代码：

interface SeatGeekApi {
  @GET("events") Observable<EventResponse> getUpcomingEvents();
}
  
Retrofit mRetrofit = new Retrofit.Builder()
    .baseUrl("https://api.seatgeek.com/")
    .build();
  
SeatGeekApi mApi = mRetrofit.create(SeatGeekApi.class);
  
mApi.getUpComingEvents()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(events -> {
      // handle events
    }, error -> {
      // handle errors
    });

上面的代码缺少了哪些东西？

• 请求失败时需要 retry
• 处理特定的 响应码
• UI 中显示 loading indicators
• 对系统 configuration changes 的响应
• 其他特殊事件

I. 重试请求 Retrying Request

网络请求在移动设备上会因各种原因发生错误，通常我们需要添加一些重试请求的机制。

一种 naive 的实现：

private void makeRequest() {
  getRequestObservable()
      .subscribe(getObserver());
}
  
private Observer<Response> getObserver() {
  return new Observer<Response>() {
    ...
    @Override public void onError(Throwable e) {
      // 发生错误时发送新的请求
      if (someCondition) {
        makeRequest();
      }
    }
  };
}

使用 retryWhen() 实现：

RxJava 的 retryWhen() 方法可以便捷地实现这个需求：

• retryWhen() 只会在每个 subscription 中调用一次
• retryWhen() 返回的是 parent 数据流的 errors
• 只要从 retryWhen() 返回的 Observable 对象不是 complete 或者 error 的，parent Observable 将会被重新订阅。

// 简单的演示重试 3 次请求，分别延迟 5s，10s，15s.
getRequestObservable()
  .retryWhen(attempt -> {
    attempt
        .zipWith(Observable.range(1, 3), (n, i) -> i)
        .flatMap(i -> {
          return Observable.timer(5 * i, TimeUnit.SECONDS);
        })
  })
  .subscribe(viewModel -> {
    // handle updated request state
  });

一些优化：

• 频繁的自动 retry 可能会导致 app DDoS 服务器。
• 比较好的处理方式是让用户选择 retry，我们可以使用 RxRelay 中的Relay/Subject。

PublishRelay<Long> retryRequest = PublishRelay.create();
    
getRequestObservable()
    .retryWhen(attempt ->retryRequest)
    .subscribe(viewModel -> {
      // handle updated request state
    });

@OnClick(R.id.retry_view)
public void onRetryClicked() {
  retryRequest.call(System.currentTimeMillis);
}

II. 处理特殊响应码 Response Codes

使用 Response<T>

• 利用 Observable<Response<T>> 可以使用响应的元数据 (metadata)。
• 这种方法的优点是服务器响应码在 400-500 之间不会产生异常。

interface SeatGeekApi {
  @GET("events") Observable<Response<EventResponse>> getUpComingEvents();
}
  
Retrofit mRetrofit = new Retrofit.Builder()
    .baseUrl("https://api.seatgeek.com/")
    .build();

mApi.getUpComingEvents()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(eventsResponse -> {
      int responseCode = eventsResponse.code();
      switch (responseCode) {
        HTTP_301:
          ...
        HTTP_403:
          ...
      }
    }, error -> {
      // ONLY handle i/o errors
    });

一些优化

• 仅仅使用 Response<T> 处理所有的 response code
  可能会显得比较累赘，幸运的是 Retrofit 提供了 HttpException 处理 errors
• 可以在 onError 回调里检查 exception 是否为 HttpException

mApi.getUpComingEvents()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(eventsResponse -> {
      // handle something with events
    }, error -> {
      if (error instanceof HttpException) {
        Response response = ((HttpException)error).response();
        switch (response.code()) {...}
      } else {
        // handle other errors
      }
    });

• 为了增强代码可读性，我们还可以使用 share().

interface SeatGeekApi {
  @GET("events") Observable<Response<EventResponse>> getUpComingEvents();
}

Observable<Response<EventResponse>> eventsResponse = mApi.getUpcomingEvents()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .share();

eventsResponse
    .filter(Response::isSuccessful)
    .subscribe(this::handleSuccessfulResponse);

eventsResponse
    .filter(response -> response.code() == HTTP_403)
    .subscribe(this::handle403Response);

eventsResponse
    .filter(response -> response.code() == HTTP_304)
    .subscribe(this::handle304Response);

III. 显示进度条 Loading Indicator

一种 naive 的实现

mApi.getUpcomingEvents()
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .doOnSubscribe(() -> loadingIndicator.show())
  .doOnUnsubscribe(() -> loadingIndicator.hide())
  .subscribe(events -> {
    // do something with events
  }, error -> {
    // handle errors
  });

上例中存在的问题：

• UI 跟数据流深度耦合
• 无法响应数据流的变化，仅在简单的请求时有效

使用分割数据流 Splitting Stream

• 尝试新的方法，UI 更新位于另一条 stream
• 我们可以灵活的根据 loading 状态更新 UI.

enum RequestState {
  IDLE, LOADING, COMPLETE, ERROR
}

BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE);

void publishRequestState(RequestState requestState) {
  Observable.just(requestState)
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(state);
}

mApi.getUpcomingEvents()
  .doOnSubscribe(() -> publishRequestState(RequestState.LOADING))
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .doOnError(t -> publishRequestState(RequestState.ERROR))
  .doOnComplete(() -> publishRequestState(RequestState.COMPLETE))
  .subscribe(events -> {
    // do something with events
  }, error -> {
    // handle errors
  });

• 请求状态发生变化时，在另一条 stream 修改 UI 的变化：

state.subscribe(requestState -> {
    switch (requestState) {
      IDLE:
        break;
      LOADING:
        loadingIndicator.show();
        errorView.hide();
        break;
      COMPLETE:
        loadingIndicator.hide();
        break;
      ERROR:
        loadingIndicator.hide();
        errorView.show();
        break;
    }
  });

一些优化

• 我们可以将这种模式扩展到各种 request result 上

BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE);
BehaviorRelay<EventsResponse> response = BehaviorRelay.create();
BehaviorRelay<Throwable> errors = BehaviorRelay.create();
...
void executeRequest() {
  mApi.getUpcomingEvents()
      .doOnSubscribe(() -> publishRequestState(RequestState.LOADING))
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .doOnError(t -> publishRequestState(RequestState.ERROR))
      .doOnComplete(() -> publishRequestState(RequestState.COMPLETE))
      .subscribe(response, errors);
  }

• 我们同样可以将所有分割的 streams 合并成一条:

BehaviorRelay<RequestState> state = BehaviorRelay.create(RequestState.IDLE);
BehaviorRelay<Optional<EventsResponse>> response = BehaviorRelay.create(Optional.empty());
BehaviorRelay<Optional<Throwable>> errors = BehaviorRelay.create(Optional.empty());
  
class RequestViewModel {
  public final RequestState mState;
  public final Optional<EventsResponse> mResponse;
  public final Optional<Throwable> mErrors;
  
  RequestViewModel(...) {...}
}

Observable.combineLatest(state, response, errors, RequestViewModel::new)
  .subscribe(viewModel -> {
    // handle updated request state
  });

IV. 管理 Configuration Changes

RxJava，以及像 AsyncTask 等其他异步模型默认情况下不会响应 Android 的生命周期。RxJava 提供了数据流 unsubscribe 的方法，但是问题是该如何正确的使用它。

Disposable mDisposable = Observable.combineLatest(state, response, errors, RequestViewModel::new)
  .subscribe(viewModel -> {
    // handle updated request state
  });

...

mDisposable.dispose(); // when?

一种基本的实现

创建 Disposable 时保存 Disposable 的引用，在生命周期发生回调时调用 dispose() 方法。

class MyActivity extends Activity {
  Disposable mDisposable;

  @Override protected void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    ...
    mDisposable = Observable
        .combineLatest(state, response, errors, ViewModel::new)
        .subscribe(getSubscriber());
  }

  @Override protected void onStop() {
    super.onStop();
    mDisposable.dispose();
  }
}

CompositeDisposable

• CompositeDisposable 非常适合 UI 订阅了多条 stream 的场景。
• 使用 add() 方法添加每条 stream 的 Disposable 对象。
• CompositeDisposable 的清理方法非常适合在 MVP 框架下的 unbind() 方法使用。

class MyActivity extends Activity {
  CompositeDisposable mDisposables = new CompositeDisposable();

  @Override protected void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    ...    
    mDisposables.add(state.subscribe(getStateSubscriber()));
    mDisposables.add(state.subscribe(getStateSubscriber()));
    mDisposables.add(state.subscribe(getStateSubscriber()));
  }

  @Override protected void onStop() {
    super.onStop();
    mDisposables.clear();
  }
}

推荐的方式—— RxLifecycle

• RxLifecycle 提供了便捷的方式将我们的 Disposable 与 Activity/Fragment 的生命周期绑定。
• 当相应的生命周期回调发生时，Disposable 将会调用 dispose() 方法。

class MyActivity extends RxAppCompatActivity {
  @Override protected void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    ...
    Observable.combineLatest(state, response, errors, ViewModel::new)
        .compose(bindToLifecycle())
        .subscribe(viewModel -> {
          // handle updated request state
        });

RxLifecycle + MVP

public class RxPresenter<T extends MvpView> extends Presenter<T> {

  BehaviorRelay<PresenterLifecycle> mPresenterLifecycle =
      BehaviorRelay.create(STOPPED);

  @Override public void attach(T mvpView) {
    mPresenterLifecycle.call(STARTED);
  }

  @Override public void detach() {
    mPresenterLifecycle.call(STOPPED);
  }

  protected <T>LifecycleTransformer<T> bindToLifecycle() {
    return RxLifecycle.bind(presenterLifecycleObservable, lifecycle -> {
      switch (lifecycle) {
        case STARTED:
          return STOPPED;
        case STOPPED:
        default:
          return STARTED;
      }
    });
  }
}