使用RxJava处理回调

Question

我想了解什么是使用RxJava处理特定情况的最佳方法。

我需要返回Observable，它处理从Android AccountManager中删除帐户。不幸的是，这个动作是异步的，它需要一个回调来处理结果。

正因为如此，我现在用的是Observable.create方法，用这种方法：

return Observable.create(subscriber -> { 
     final Account account = getAccountIfExist(); 
     if (account == null) { 
      subscriber.onError(new NoUserFoundException()); 
      return; 
     } 

     accountManager.removeAccount(account, future -> { 
      boolean result = false; 
      try { 
       result = future.getResult(); 
      } catch (Exception ex) { 
       Log.e(TAG, "Remove account not successful : ", ex); 
       subscriber.onError(ex); 
       return; 
      } 

      if (result) { 
       subscriber.onNext(account.name); 
       subscriber.onCompleted(); 
      } else { 
       subscriber.onError(new RuntimeException("Cannot remove the account.")); 
      } 
     }, null); 
    }); 

但是，Observable.create有一个处理背压和消除不同的问题，如本post

所以，问题说明是，我怎样才能使用Observable.fromCallable（或等效的方法）来处理回调（一般情况下）以及处理背压和取消？

Answer 1

你可以尝试使用

Observable.defer

return Observable.defer(() -> { 
    final Account account = getAccountIfExist(); 
     if (account == null) { 
      return Observable.error(new NoUserFoundException()); 
     } 
accountManager.removeAccount(account, future -> { 
      boolean result = false; 
      try { 
       result = future.getResult(); 
      } catch (Exception ex) { 
       return Observable.error(ex); 
      } 

      if (result) { 
       Observable.just(account.name); 
      } else { 
       return Observable.error(new NoUserFoundException("Cannot remove the account.")) 
      } 
     }, null); 
});