Xử lý các ngoại lệ API trong RxJava

Question

Tôi đang cố gắng bọc đầu xung quanh RxJava, nhưng tôi gặp một chút rắc rối khi xử lý các ngoại lệ gọi dịch vụ một cách thanh lịch.

Về cơ bản, tôi có dịch vụ (trang bị thêm) trả lại số Observable<ServiceResponse>. ServiceResponse được định nghĩa như sau:

public class ServiceResponse { 
    private int status; 
    private String message; 
    private JsonElement data; 

    public JsonElement getData() { 
     return data; 
    } 

    public int getStatus() { 
     return status; 
    } 

    public String getMessage() { 
     return message; 
    } 
} 

Bây giờ những gì tôi muốn là để lập bản đồ rằng phản ứng tổng quát để một List<Account> chứa trong các lĩnh vực dữ liệu JsonElement (tôi giả sử bạn không quan tâm những gì các đối tượng Account trông như thế nào, vì vậy tôi đã thắng 'gây ô nhiễm bài đăng với nó). Mã sau hoạt động thực sự tốt cho trường hợp thành công, nhưng tôi không thể tìm thấy cách tốt để xử lý các ngoại lệ API của mình:

service.getAccounts() 
     .subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .map(new Func1<ServiceResponse, AccountData>() { 
       @Override 
       public AccountData call(ServiceResponse serviceResponse) { 

        // TODO: ick. fix this. there must be a better way... 
        ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus()); 
        switch (responseType) { 
         case SUCCESS: 
          Gson gson = new GsonBuilder().create(); 
          return gson.fromJson(serviceResponse.getData(), AccountData.class); 
         case HOST_UNAVAILABLE: 
          throw new HostUnavailableException(serviceResponse.getMessage()); 
         case SUSPENDED_USER: 
          throw new SuspendedUserException(serviceResponse.getMessage()); 
         case SYSTEM_ERROR: 
         case UNKNOWN: 
         default: 
          throw new SystemErrorException(serviceResponse.getMessage()); 
        } 
       } 
     }) 
     .map(new Func1<AccountData, List<Account>>() { 
       @Override 
       public List<Account> call(AccountData accountData) { 
        Gson gson = new GsonBuilder().create(); 
        List<Account> res = new ArrayList<Account>(); 
        for (JsonElement account : accountData.getAccounts()) { 
         res.add(gson.fromJson(account, Account.class)); 
        } 
        return res; 
       } 
     }) 
     .subscribe(accountsRequest); 

Có cách nào tốt hơn không? không hoạt động, lỗi này sẽ kích hoạt người quan sát của tôi và tôi sẽ nhận được lỗi mà tôi đã ném, nhưng chắc chắn dường như tôi không làm điều này đúng.

Cảm ơn trước!

Edit:

Hãy để tôi làm rõ chính xác những gì tôi muốn đạt được:

Tôi muốn có một lớp học mà có thể được gọi từ UI (ví dụ như một hoạt động, hoặc Fragment, hoặc bất cứ điều gì) . lớp đó sẽ mất một Observer<List<Account>> như một tham số như sau:

public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) { 
    ... 
} 

rằng phương pháp sẽ trả về một thuê bao thể huỷ đăng ký khi giao diện người dùng được tách/phá hủy/etc.

Trình quan sát được tham số hóa sẽ xử lý onNext cho các câu trả lời thành công trong danh sách Tài khoản. OnError sẽ xử lý bất kỳ ngoại lệ nào, nhưng cũng sẽ vượt qua bất kỳ ngoại lệ API nào (ví dụ: nếu trạng thái phản hồi! = 200, chúng tôi sẽ tạo Throwable và chuyển nó lên onError). Lý tưởng nhất là tôi không muốn chỉ "ném" ngoại lệ, tôi muốn chuyển nó trực tiếp cho người quan sát. Đó là tất cả những gì tôi thấy.

Biến chứng là dịch vụ Retrofit của tôi trả về đối tượng ServiceResponse, vì vậy người quan sát của tôi không thể đăng ký điều đó. Điều tốt nhất tôi đã đi lên với là tạo ra một wrapper Observer xung quanh Observer của tôi, như vậy:

@Singleton 
public class AccountsDatabase { 

    private AccountsService service; 

    private List<Account> accountsCache = null; 
    private PublishSubject<ServiceResponse> accountsRequest = null; 

    @Inject 
    public AccountsDatabase(AccountsService service) { 
     this.service = service; 
    } 

    public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) { 

     ObserverWrapper observerWrapper = new ObserverWrapper(observer); 

     if (accountsCache != null) { 
      // We have a cached value. Emit it immediately. 
      observer.onNext(accountsCache); 
     } 

     if (accountsRequest != null) { 
      // There's an in-flight network request for this section already. Join it. 
      return accountsRequest.subscribe(observerWrapper); 
     } 

     if (accountsCache != null && !forceRefresh) { 
      // We had a cached value and don't want to force a refresh on the data. Just 
      // return an empty subscription 
      observer.onCompleted(); 
      return Subscriptions.empty(); 
     } 

     accountsRequest = PublishSubject.create(); 

     accountsRequest.subscribe(new ObserverWrapper(new EndObserver<List<Account>>() { 

      @Override 
      public void onNext(List<Account> accounts) { 
       accountsCache = accounts; 
      } 

      @Override 
      public void onEnd() { 
       accountsRequest = null; 
      } 
     })); 

     Subscription subscription = accountsRequest.subscribe(observerWrapper); 

     service.getAccounts() 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(accountsRequest); 

     return subscription; 
    } 

    static class ObserverWrapper implements Observer<ServiceResponse> { 

     private Observer<List<Account>> observer; 

     public ObserverWrapper(Observer<List<Account>> observer) { 
      this.observer = observer; 
     } 

     @Override 
     public void onCompleted() { 
      observer.onCompleted(); 
     } 

     @Override 
     public void onError(Throwable e) { 
      observer.onError(e); 
     } 

     @Override 
     public void onNext(ServiceResponse serviceResponse) { 
      ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus()); 
      switch (responseType) { 
       case SUCCESS: 
        Gson gson = new GsonBuilder().create(); 
        AccountData accountData = gson.fromJson(serviceResponse.getData(), AccountData.class); 
        List<Account> res = new ArrayList<>(); 
        for (JsonElement account : accountData.getAccounts()) { 
         res.add(gson.fromJson(account, Account.class)); 
        } 
        observer.onNext(res); 
        observer.onCompleted(); 
        break; 
       default: 
        observer.onError(new ApiException(serviceResponse.getMessage(), responseType)); 
        break; 
      } 
     } 
    } 
} 

tôi vẫn cảm thấy như tôi không sử dụng này một cách chính xác mặc dù. Tôi chắc chắn đã không nhìn thấy bất cứ ai khác bằng cách sử dụng ObserverWrapper trước đây. Có lẽ tôi không nên sử dụng RxJava, mặc dù những người ở SoundCloud và Netflix thực sự đã bán tôi trên đó trong bài thuyết trình của họ và tôi rất háo hức muốn học nó.

Answer 1

Vui lòng đọc bên dưới Tôi đã thêm chỉnh sửa.

Hoàn toàn chính xác để ném trong Hành động/Func/Observer với RxJava. Ngoại lệ sẽ được tuyên truyền bởi khung bên phải xuống đến Observer của bạn. Nếu bạn giới hạn mình chỉ gọi onError thì bạn sẽ tự xoay mình để thực hiện điều đó.

Với điều đó được cho là một đề xuất sẽ chỉ đơn giản là loại bỏ trình bao bọc này và thêm xác thực đơn giản Hành động trong chuỗi dịch vụ Observ.getAccount ....

Tôi muốn sử dụng doOnNext (mới ValidateServiceResponseOrThrow) bị xích với một bản đồ (MapValidResponseToAccountList mới). Đó là những lớp đơn giản thực hiện mã cần thiết để giữ cho chuỗi Quan sát dễ đọc hơn một chút.

Đây là phương pháp tải tài khoản của bạn được đơn giản hóa bằng những gì tôi đã đề xuất.

public Subscription loadAccounts(Observer<List<Account>> observer, boolean forceRefresh) { 
    if (accountsCache != null) { 
     // We have a cached value. Emit it immediately. 
     observer.onNext(accountsCache); 
    } 

    if (accountsRequest != null) { 
     // There's an in-flight network request for this section already. Join it. 
     return accountsRequest.subscribe(observer); 
    } 

    if (accountsCache != null && !forceRefresh) { 
     // We had a cached value and don't want to force a refresh on the data. Just 
     // return an empty subscription 
     observer.onCompleted(); 
     return Subscriptions.empty(); 
    } 

    accountsRequest = PublishSubject.create(); 
    accountsRequest.subscribe(new EndObserver<List<Account>>() { 

     @Override 
     public void onNext(List<Account> accounts) { 
      accountsCache = accounts; 
     } 

     @Override 
     public void onEnd() { 
      accountsRequest = null; 
     } 
    }); 

    Subscription subscription = accountsRequest.subscribe(observer); 

    service.getAccounts() 
      .doOnNext(new ValidateServiceResponseOrThrow()) 
      .map(new MapValidResponseToAccountList()) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(accountsRequest); 

    return subscription; 
} 

private static class ValidateResponseOrThrow implements Action1<ServiceResponse> { 
     @Override 
     public void call(ServiceResponse response) { 
      ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus()); 
      if (responseType != SUCCESS) 
       throw new ApiException(serviceResponse.getMessage(), responseType)); 
     } 
    } 

private static class MapValidResponseToAccountList implements Func1<ServiceResponse, List<Account>> { 
    @Override 
    public Message call(ServiceResponse response) { 
     // add code here to map the ServiceResponse into the List<Accounts> as you've provided already 
    } 
} 

Edit: Trừ khi ai đó nói nếu không tôi nghĩ đó là cách tốt nhất để trở lại các lỗi sử dụng flatMap. Tôi đã gửi Ngoại lệ từ Hành động trong quá khứ nhưng tôi không tin đó là cách được đề xuất.

Bạn sẽ có ngăn xếp Ngoại lệ sạch hơn nếu bạn sử dụng flatMap. Nếu bạn ném từ bên trong ngăn xếp Hành động, ngoại lệ sẽ thực sự chứa rx.exceptions.OnErrorThrowable$OnNextValue Ngoại lệ không lý tưởng.

Để tôi minh họa ví dụ trên bằng cách sử dụng flatMap thay thế.

private static class ValidateServiceResponse implements rx.functions.Func1<ServiceResponse, Observable<ServiceResponse>> { 
    @Override 
    public Observable<ServiceResponse> call(ServiceResponse response) { 
     ResponseTypes responseType = ResponseTypes.from(serviceResponse.getStatus()); 
     if (responseType != SUCCESS) 
      return Observable.error(new ApiException(serviceResponse.getMessage(), responseType)); 
     return Observable.just(response); 
    } 
} 

service.getAccounts() 
    .flatMap(new ValidateServiceResponse()) 
    .map(new MapValidResponseToAccountList()) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(accountsRequest); 

Như bạn có thể thấy sự khác biệt là tinh tế. ValidateServiceResponse hiện đang triển khai Func1 thay vì Action1 và chúng tôi không còn sử dụng từ khóa throw nữa. Thay vào đó, chúng tôi sử dụng . Tôi tin rằng điều này phù hợp hơn với hợp đồng Rx dự kiến.

Answer 2

Bạn có thể đọc bài viết này tốt về xử lý http://blog.danlew.net/2015/12/08/error-handling-in-rxjava/