取消長時間運行的任務內部Observable內置訂閱處置

Question

我正在寫一個被動的封裝在凌亂的庫上以輕鬆地在我的應用中發送http請求。這裏是類：

/** 
* Used to send a http GET/POST request. 
*/ 

public class BasicRequest { 
public static final String LOG_TAG = "BasicRequest"; 

public static final int GET_REQUEST = Request.Method.GET; 
public static final int POST_REQUEST = Request.Method.POST; 
private final int mRequestType; 
private final String mServiceLocation; 
private final Map<String, String> mParams; 

/** 
* Keeps track of all the request for this object. Will be helpful when we need to cancel 
* the request when someone disposes the subscription. 
*/ 
private List<StringRequest> mStringRequests = new ArrayList<>(); 

private Context mContext; 

private int mRequestTimeout = BASIC_REQUEST_DEFAULT_TIMEOUT; 

public BasicRequest(Context context, 
        String serviceLocation, 
        int requestType, 
        final Map<String, String> params) { 
    mContext = context; 
    mRequestType = requestType; 
    mServiceLocation = serviceLocation; 
    mParams = params; 
} 

private void fireRequest(final SingleEmitter<String> e) { 

    StringRequest stringRequest; 
    if(mRequestType == GET_REQUEST) { 
     stringRequest = new StringRequest(Request.Method.GET, mServiceLocation, 
       new Response.Listener<String>() { 
        @Override 
        public void onResponse(String response) { 
         e.onSuccess(response); 

        } 
       }, new Response.ErrorListener() { 
      @Override 
      public void onErrorResponse(VolleyError error) { 
       e.onError(error); 
      } 
     }); 
    } else { 
     stringRequest = new StringRequest(Request.Method.POST, mServiceLocation, 
       new Response.Listener<String>() { 
        @Override 
        public void onResponse(String response) { 
         e.onSuccess(response); 
        } 
       }, new Response.ErrorListener() { 
      @Override 
      public void onErrorResponse(VolleyError error) { 
       e.onError(error); 
      } 
     }) { 
      @Override 
      protected Map<String, String> getParams() throws AuthFailureError { 
       return mParams; 
      } 

     }; 
    } 
    mStringRequests.add(stringRequest); 
    stringRequest.setRetryPolicy(new DefaultRetryPolicy(
      mRequestTimeout, 
      ConnectionUtils.BASIC_REQUEST_DEFAULT_RETRIES, 
      DefaultRetryPolicy.DEFAULT_BACKOFF_MULT)); 
    VolleyInstance.getInstance(mContext).addToRequestQueue(stringRequest); 
} 


/** 
* Returns a Single observable for results. Queues the request on Subscription. Must be 
* called only once during the lifetime of object. Calling multiple times will return null. 
* Expect to get VolleyException in case of error. 
* @return Single observable for String results. If it's is used for second time, it will 
* return null. 
*/ 
@Nullable 
public Single<String> get() { 

    return Single.create(new SingleOnSubscribe<String>() { 
     @Override 
     public void subscribe(@NonNull SingleEmitter<String> e) throws Exception { 
      fireRequest(e); 
     } 
    }).doOnDispose(new Action() { 
     @Override 
     public void run() throws Exception { 
      for (StringRequest stringRequest: mStringRequests) { 
       stringRequest.cancel(); 
      } 
     } 
    }); 

} 

/** 
* Set the request timeout for this request. 
* @param requestTimeout time in milliseconds. 
*/ 
public void setRequestTimeout(int requestTimeout) { 
    mRequestTimeout = requestTimeout; 
} 

現在的問題是，當有人處置訂閱所有相應的請求將被停止。有沒有辦法我只能停止處理訂閱的請求？

我知道一旦實現它將強制只有一個訂閱可以維護，如果有人再次調用get，cache'd觀察者將被返回。基於訂閱處置是否有更好的方式處理HTTP請求？

Answer 1

你不需要去管理外fireRequest，SingleEmitter有setCancellable方法正好爲，做到消除那裏，RxJava將確保調用它，當有人處置觀察到。

添加在fireRequest()方法，並刪除doOnDispose：

e.setCancellable(()-> stringRequest.cancel());