RxJava练武场之——基于RxJava Retrofit网络框架的搭建

RxJava练武场是一个rxjava在项目中应用的小系列，包括：

• RxJava练武场之——基于Observable网络框架的搭建
• RxJava练武场之——Observable网络框架的解耦和复用
• RxJava练武场之——Token前置请求

Observable网络框架的抽象

Observable网络框架建立的原因

两个问题：

• Retrofit已经对网络请求做了封装，为什么还要封装?

  网络请求中对于请求流程、配置、加解密、异常处理每个app都是固定不变的，如果业务每次请求都自己处理，会存在冗余代码，且质量不易保证。所以我们需要基于Retrofit对这些流程和操作做二次封装，并对调用方式进行统一，我们称之为网络框架。

• 框架封装方式Observable是什么？

  网络框架，传统使用callback/listener方式异步回调网络请求结果。但是这种callback的方式，没有利用到Retrofit的一大优势--rxjava调用，所以我们要基于rxjava调用方式，封装一个基于Observable的网络请求框架。 以下所说网络框架，均指基于Observable的网络请求二次封装框架。

Observable网络框架设计目标

1. T1：支持Get、Post请求，对Request的业务params、url可轻松配置
2. T2：对Request中params组合和加密，统一处理
3. T3：返回Response 解密处理，并自动JavaBean化
4. T4：返回Response code/status 判断及统一处理
5. T5：网络请求cancle机制、progressBar等通用处理

Observable网络框架如何实现

设计原则：

1. 网络框架Api返回Observable对象，并作为网络请求事件的生产者： 生产者负责请求的发起，和返回数据的处理。

2. 业务注册Observer类，作为消费者。

   只负责对返回数据的响应

生产者和消费者界限明确，做到完全解耦，为网络请求与其他rxjava异步调用的整合打基础。

这篇文章详细讲述框架中生产者的实现，消费者的实现在《基于Rxjava Retrofit的网络框架（二）》中介绍

在Observable的创建过程中，框架如何封装？

首先我们需要一个Manager或Helper全局句柄，通过他可以发起网络请求，一般设计为单例全局持有，有利于网络请求一些资源的共用。 我们暂定为NetHelper,其网络请求接口定义为：

private static <R extends HttpResponse<?>> Observable sendRequest(final ObservableRequest<R> orgRequest) 
复制代码

sendRequest方法中，我们来看下Observable对象的生成过程：参考Retrofit本身Observable生成方式

第一步

定义Request，Request类的定义在Retrofit里通过注解的方式完成的

public interface Request {

    @POST("{url}")
    Observable<JSONObject> postJSONResult(@Path(value="url",encoded =true) String url, @FieldMap Map<String, String> params);
}

复制代码

原来Retrofit中每个请求都要为request定义一个interface，并且返回的response类型需要此处指明，那除此之外请求的入参要在另一个地方单独定义。

我们定义的Observable泛型是通用的JSONObject，这样做的好处是每个Request子类，只需要一个类声明自己入参和url，无需单独定义interface声明response类型。

第二步

创建retrofit；

NetHelper.java中

// 初始化okhttp
OkHttpClient client = new OkHttpClient.Builder()
        .build();
/**
 *OkHttpClient每次请求的时候都要创建，注意：OkHttpClient.Builder()中有ConnectionPool作为OkHttp的   	*连接池要复用，否则请求过多时容易导致内存溢出
**/

// 初始化Retrofit
retrofit = new Retrofit.Builder()
        .client(client)
        .baseUrl(Request.HOST)
        .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
        .addConverterFactory(MyConverterFactory.create())
        .build();

复制代码

一般addConverterFactory配置是GsonConverterFactory，用于把Response通过GSon转为javaBean（Request接口中定义的类型），由于我们定义为了JsonObject，所以我们使用MyConverterFactory自定义实现

第三步

生成Observable

Observable的真正创建，我们放到Request基类中

public abstract class HttpRequest<T extends HttpResponse>{
    protected abstract String getURLAction();
    
    //Observable的创建
    public Observable getObservable(Retrofit retrofit) {
        Request request = retrofit.create(Request.class);
		return request.postJSONResult(getURLAction(),getURLParam());
    }
    
    public String getBaseURL(){
        //return base url;
    }
    
    public SortedMap<String, String> getURLParam() {
    	//return 对HttpRequest子类(业务类)，定义的params，进行组合和加密
    	//通用的组合和加密逻辑就在此处。
    }
    
    //response类型解析
    protected Type type;

    public ObservableMapiRequest(BaseContext context) {
        super(context);
        initType();
    }

    private void initType() {
        Type superClass = getClass().getGenericSuperclass();

        this.type = ((ParameterizedType) superClass).getActualTypeArguments()[0];

    }

    public Type getType() {
        return type;
    }
}
复制代码

以上三步，已经初步将Observable返回。通过以上几步只是基于Retrofit自身的Observable创建方法做了一些封装。下面的处理是框架的重点和核心：

private static <R extends HttpResponse<?>> Observable sendRequest(final ObservableRequest<R> orgRequest){
        return NetHelper.getApiObservable(orgRequest)
                .map(new JavaBeanFunc(orgRequest.getType()))
                .filter(new LogInterceptor(orgRequest))
                .compose(ResponseTransformer.handleResult())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());

    }
复制代码

NetHelper.getApiObservable方法后，再加上网络请求的线程配置，这时候业务subscribe消费者，就可以直接得到解密后的JsonObject了。注意此时只是JSONObject，我们需要转换成我们需要的Response。

我们在HttpRequest类中看到了，定义HttpRequest子类时，是需要传入Response类型的（就是T），在HttpRequest类中已经将Response的类型解析出来，并保存了。

public class JavaBeanFunc<J extends JSONObject,T extends HttpResponse> implements Function<J,T> {
    Type mCon;
    public JavaBeanFunc(Type con){
        mCon = con;
    }
    @Override
    public T apply(J jsonObject) throws Exception {
        if (jsonObject != null){
            T response = jsonObject.toJavaObject(mCon);
            return response;
        } else {
            return null;
        }
    }
}
复制代码

同时配置了拦截器，用于log输出，和异常的拦截,也可以做网络性能指标的记录等

public class LogInterceptor<R extends HttpResponse<?>> implements Predicate<R> {
    private HttpRequest<R> orgRequest;

    public LogInterceptor(HttpRequest<R> request){
        orgRequest = request;
    }

    @Override
    public boolean test(R response) throws Exception {
        boolean isPredicate = false;
        if (orgRequest != null && response != null) {
            HttpHelper.printHttpLog(orgRequest, JSONObject.toJSONString(response));
            isPredicate = true;
        }
        return isPredicate;
    }
}
复制代码

response解析完毕之后，还没有完事；必须对response的正常和异常两种情况做一个判断，哪些情况作为onNext传递，哪些情况作为onError传递，也要定义清楚。

有些文章对于progressbar的控制也放到这里，我认为并不符合框架的设计原则，不属于生产者该做的事情。

public static <T extends Serializable> ObservableTransformer<T, T> handleResult() {
    return upstream -> upstream
            .onErrorResumeNext(new ErrorResumeFunction<T>())
            .flatMap(new ResponseFunction<T>());
}

private static class ErrorResumeFunction<T extends Serializable> implements Function<Throwable, ObservableSource<? extends T>> {

    @Override
    public ObservableSource<? extends T> apply(Throwable throwable) throws Exception {
        return Observable.error(CustomException.handleException(throwable));
    }
}

private static class ResponseFunction<T extends Serializable> implements Function<T, ObservableSource<T>> {

    @Override
    public ObservableSource<T> apply(T tResponse) throws Exception {
        int code = tResponse.getCode();
        String message = tResponse.getMsg();

        if (code == SUCCESS.value()) {
            return Observable.just(tResponse);
        } else {
            return Observable.error(new ApiException(code, message));
        }
    }
}
复制代码

你可能有两个疑问，一个是response中code的判定可以在observer中处理吗，另一个是服务器错误和业务错误为何都作为error抛出。

第一个问题： code值的判定不可以在observer中处理，而必须在Observable一端处理。因为Observable形式的网络请求是作为数据流中的一环出现的，可能当前网络请求只是一连串异步调用（rxjava调用）的一环。 第二个问题： response中code!=SUCCESS是业务错误的情况，必须向数据流中发出，让业务处理此异常。（那同时对于Response的定义也是，code!=SUCCESS必须是不需要业务处理的情况才行） 两种错误都抛出error（内部code不同），方便架构使用者在事件响应时，既能捕捉所有错误，又能区分错误的类型。