android – 如何使用RxJava Observer而不是Event Bus处理401,403,503,500等HTTP错误
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了android – 如何使用RxJava Observer而不是Event Bus处理401,403,503,500等HTTP错误，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含11075字，纯文字阅读大概需要16分钟。
内容图文
我正在使用Retrofit,OK-HTTP和RxJava2来处理网络调用,我创建了下面的拦截器来处理每个网络调用的网络错误响应,有没有更好的方法来处理它？
这是EventBus的情况吗？
我不想在每种方法中检查此错误异常,
// HTTP客户端
OkHttpClient tempClient = new OkHttpClient.Builder()
.readTimeout(CONNECT_TIMEOUT_IN_SEC, TimeUnit.SECONDS)// connect timeout
.connectTimeout(CONNECT_TIMEOUT_IN_SEC, TimeUnit.SECONDS)// socket timeout
.followRedirects(false)
.cache(provideHttpCache())
.addNetworkInterceptor(new ResponseCodeCheckInterceptor())
.addNetworkInterceptor(new ResponseCacheInterceptor())
.addInterceptor(new AddHeaderAndCookieInterceptor())
.build();
HTTP客户端拦截器
public class ResponseCodeCheckInterceptor implements Interceptor {
private static final String TAG = "RespCacheInterceptor";
@Override
public Response intercept(Chain chain) throws IOException {
Response response = chain.proceed(chain.request());
Request originalRequest = chain.request();
if (response.code() == HttpStatus.UNAUTHORIZED.value()) {
throw new UnAuthorizedException();
}else if (response.code() == HttpStatus.INTERNAL_SERVER_ERROR.value()) {
throw new APIException(response.code(), "Server Internal Error");
} else if (response.code() == HttpStatus.SERVICE_UNAVAILABLE.value()) {
throw new ServiceUnavailableException();
} else {
throw new APIException(code,response.body().toString());
}
return response;
}
}
API类
@GET("customer/account/")
Single<Customer> getCustomer();
......
存储库类
@Override
public Single<Customer> getCustomer() {
return this.mCustomerRemoteDataStore.getCustomer()
.doOnSuccess(new Consumer<Customer>() {
@Override
public void accept(Customer customer) throws Exception {
if (customer != null) {
mCustomerLocalDataStore.saveCustomer(customer);
}
}
}).doOnError(new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
}
});
}
演讲者课程
@Override
public void getCustomerFullDetails() {
checkViewAttached();
getView().showLoading();
addSubscription(customerRepository.getCustomerFullDetails(true)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<CustomerDetails>() {
@Override
public void onSuccess(@io.reactivex.annotations.NonNull CustomerDetails customerDetails) {
getView().onCustomerDetailsSuccess();
}
@Override
public void one rror(@io.reactivex.annotations.NonNull Throwable throwable) {
Log.d(TAG, "error: " + throwable.getLocalizedMessage());
if (throwable instanceof UnAuthorizedException) {
getView().showLoginPage();
else if (throwable instanceof ServiceUnavailableException) {
getView().showServiceUnAvaiableMsg();
}...
}
})
);
}
更新的代码
============
public class CheckConnectivityInterceptor implements Interceptor {
private static final String TAG = CheckConnectivityInterceptor.class.getSimpleName() ;
private boolean isNetworkActive;
private RxEventBus eventBus;
private Context mContext;
public CheckConnectivityInterceptor(RxEventBus eventBus, Context mContext) {
this.mContext = mContext;
this.eventBus = eventBus;
}
@Override
public Response intercept(Interceptor.Chain chain) throws IOException {
Request originalRequest = chain.request();
String requestPath = originalRequest.url().url().getPath();
if (!NetworkUtil.isConnected(this.mContext)) {
eventBus.send(new ErrorState(ErrorType.NO_INTERNET_CONNECTION,
this.mContext.getString(R.string.no_network_connection), requestPath));
//Added this exception so it's not trying to execute the chain
throw new NoConnectivityException();
} else {
Response originalResponse = null;
try {
originalResponse = chain.proceed(chain.request());
} catch (Exception ex) {
eventBus.send(new ErrorState(ErrorType.SERVICE_ERROR, this.mContext.getString(R.string.connection_failed), requestPath));
Log.e(TAG, "check connectivity intercept: ",ex );
throw new IOException("IO Exception occurred");
}
return originalResponse;
}
}
}
====================
public class HTTPResponseCodeCheckInterceptor implements Interceptor {
private RxEventBus eventBus;
public HTTPResponseCodeCheckInterceptor(RxEventBus eventBus) {
this.eventBus = eventBus;
}
@Override
public Response intercept(Chain chain) throws IOException {
if (!responseSuccess) {
if (code == HttpStatus.MOVED_TEMPORARILY.value()) {
eventBus.send(new ErrorState(ErrorType.STEP_UP_AUTHENTICATION,requestPath,rSecureCode));
} else if (code == HttpStatus.INTERNAL_SERVER_ERROR.value()) { // Error code 500
eventBus.send(new ErrorState(ErrorType.SERVICE_ERROR, getAPIError(responseStringOrig),requestPath));
} else if (code == HttpStatus.SERVICE_UNAVAILABLE.value()) {
eventBus.send(new ErrorState(ErrorType.SERVICE_UNAVAILABLE, getOutageMessage(responseStringOrig),requestPath));
} else {
eventBus.send(new ErrorState(ErrorType.SERVICE_ERROR,new APIErrorResponse(500, "Internal Server Error"),requestPath));
}
}
}
}
===================
public class RxEventBus {
private PublishSubject<ErrorState> bus = PublishSubject.create();
private RxEventBus() {
}
private static class SingletonHolder {
private static final RxEventBus INSTANCE = new RxEventBus();
}
public static RxEventBus getBus() {
return RxEventBus.SingletonHolder.INSTANCE;
}
public void send(ErrorState o) {
bus.onNext(o);
}
public Observable<ErrorState> toObserverable() {
return bus;
}
public boolean hasObservers() {
return bus.hasObservers();
}
public static void register(Object subscriber) {
//bus.register(subscriber);
}
public static void unregister(Object subscriber) {
// bus.unregister(subscriber);
}
}
=====================
public class BasePresenter<V extends MVPView> implements MVPPresenter<V> {
private final CompositeDisposable mCompositeDisposable;
@Override
public void subscribe() {
initRxBus();
}
@Override
public void unsubscribe() {
RxUtil.unsubscribe(mCompositeDisposable);
}
public void addSubscription(Disposable disposable){
if(mCompositeDisposable != null){
mCompositeDisposable.add(disposable);
Log.d(TAG, "addSubscription: "+mCompositeDisposable.size());
}
}
private void initRxBus() {
addSubscription(EPGApplication.getAppInstance().eventBus()
.toObserverable()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<ErrorState>() {
@Override
public void accept(ErrorState errorState) throws Exception {
if (mMvpView != null) {
mMvpView.hideLoadingIndicator();
if (ErrorType.STEP_UP_AUTHENTICATION == errorState.getType()) {
mMvpView.showStepUpAuthentication(errorState.getSecureRequestCode());
} else if (ErrorType.SERVICE_ERROR == errorState.getType()) {
mMvpView.showServiceError(((APIErrorResponse) errorState.getErrorData()).getErrorMessage());
} else if (ErrorType.SERVICE_UNAVAILABLE == errorState.getType()) {
mMvpView.showServiceUnavailable(((OutageBody) errorState.getErrorData()));
} else if (ErrorType.UNAUTHORIZED == errorState.getType()) {
mMvpView.sessionTokenExpiredRequestLogin();
} else if (ErrorType.GEO_BLOCK == errorState.getType()) {
mMvpView.showGeoBlockErrorMessage();
} else if (ErrorType.SESSION_EXPIRED == errorState.getType()) {
mMvpView.sessionTokenExpiredRequestLogin();
}else if (ErrorType.NO_INTERNET_CONNECTION == errorState.getType()) {
mMvpView.showNoNetworkConnectivityMessage();
mMvpView.showServiceError(resourceProvider.getString(R.string.no_network_connection));
}
}
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.e(TAG, "base excpetion: ", throwable);
}
}));
}
}
}
==================
public class ProfilePresenter<V extends ProfileContract.View> extends BasePresenter<V>
implements ProfileContract.Presenter<V> {
public ProfilePresenter(ProfileContract.View view, CustomerRepository repository) {
super();
this.repository = repository;
}
private void updateCustomerAccountDetails(JSONObject payload) {
getMvpView().showLoadingIndicator();
addSubscription(repository.updateCustomerDetails(sharedPreferencesRepository.isStepUpAuthRequired(), AppConfig.CUSTOMER_ACCOUNT_HOLDER_UPDATE
, payload)
.compose(RxUtil.applySingleSchedulers())
.subscribeWith(new DisposableSingleObserver<BaseServerResponse>() {
@Override
public void onSuccess(BaseServerResponse response) {
if (!isViewAttached()) {
return;
}
getMvpView().hideLoadingIndicator();
getMvpView().onSuccessProfileInfoUpdate();
}
@Override
public void one rror(Throwable throwable) {
if (!isViewAttached()) {
return;
}
if (throwable instanceof NoConnectivityException) {
getMvpView().showNoNetworkConnectivityMessage();
}
getMvpView().hideLoadingIndicator();
}
}));
}
}
解决方法:
我过去使用的一种方法是共享用于传达错误条件的主题.
>在生产者方面,保持对它的引用类型(发布/行为/重放/等),以便在应用程序发生的下一个错误的情况下调用onNext().在这种情况下,生产者将是ResponseCodeCheckInterceptor实例.而不是抛出各种异常,这个实例将发出一个描述刚刚发生的错误的ErrorState. (假设ErrorState是一个自定义类型,它带有关于错误条件的足够信息,供消费者决定如何做出反应,例如更新UI,清理资源等).
>在消费者方面,将对共享主题的引用保持为Observable< ErrorState>.当你似乎在做MVP时,Presenter可能会成为你的消费者之一.
(依赖注入是共享Subject实例的好方法).
希望有所帮助！
更新一些粗略的示例代码……
// this is descriptive way to identify unique error types you may care about
enum ErrorType {
Unauthorized,
ServiceUnavailable,
ServiceError
}
// this class should bundle together enough information about an
// error that has occurred so consumers can decide how to respond
// e.g. is this a "fatal" error, or can the associated operation be
// retried?
class ErrorState {
public final ErrorType type;
public final boolean fatal;
ErrorState(ErrorType type, boolean fatal) {
this.type = type;
this.fatal = fatal;
}
}
// the Interceptor creates new instances of ErrorState and pushes
// them through the Subject to notify downstream subscribers
class ResponseCodeCheckInterceptor implements Interceptor {
private final Subject<ErrorState> errorStateSubject;
ResponseCodeCheckInterceptor(Subject<ErrorState> errorStateSubject) {
this.errorStateSubject = errorStateSubject;
}
@Override
public Response intercept(@NonNull Chain chain) throws IOException {
final Response response = chain.proceed(chain.request());
if(response.code() == HttpStatus.UNAUTHORIZED.value()) {
errorStateSubject.onNext(new ErrorState(ErrorType.Unauthorized, false));
} else if (response.code() == HttpStatus.INTERNAL_SERVER_ERROR.value()) {
errorStateSubject.onNext(new ErrorState(ErrorType.ServiceError, true));
} else if (response.code() == HttpStatus.SERVICE_UNAVAILABLE.value()) {
errorStateSubject.onNext(new ErrorState(ErrorType.ServiceUnavailable, false));
}
return response;
}
}
// the Presenter is created with a reference to the same Subject, but
// retains a reference to it as an Observable. the Consumer instance
// supplied as the onNext() handler is where you'd put your logic to
// handle ErrorStates. also, the Presenter should be lifecycle-aware
// so as to create and dispose of subscriptions at the appropriate
// times.
class Presenter {
private final Observable<ErrorState> errorStateStream;
private Disposable errorStateSubscription;
class Presenter(Observable<ErrorState> errorStateStream) {
this.errorStateStream = errorStateStream
}
public void onStart() {
errorStateSubscription = errorStateStream.subscribe(
next -> {
/* Invoke views/etc */
},
error -> {
/* handle stream error during */
}
);
}
public void onStop() {
if(errorStateSubscription != null) {
errorStateSubscription.dispose();
}
}
}
内容总结
以上是互联网集市为您收集整理的android – 如何使用RxJava Observer而不是Event Bus处理401,403,503,500等HTTP错误全部内容，希望文章能够帮你解决android – 如何使用RxJava Observer而不是Event Bus处理401,403,503,500等HTTP错误所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】