使用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); 
});