Lang andauernde Aufgabe innerhalb von Observable abbrechen, wenn ein Abonnement entsorgt wird

Question

Ich schreibe einen reaktiven Wrapper über Volley-Bibliothek, um http-Anfrage einfach in meiner App zu senden. Hier ist die Klasse:

/** 
* 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; 
} 

Das Problem ist jetzt, wenn jemand ein Abonnement verfügt das alle Anforderung an alle Abonnements entsprechenden gestoppt. Gibt es einen Weg, wie ich die Anfrage nur stoppen kann, für welches Abonnement es sich handelt?

Ich weiß, sobald Weg, es zu erreichen, würde erzwingen, dass nur eine Subskription beibehalten werden kann und wenn jemand Anrufe wieder erhält, wird cache'd Beobachter zurückgegeben. Gibt es eine bessere Möglichkeit, eine http-Anfrage basierend auf einer Abonnementverfügung zu veröffentlichen?

Answer 1

Sie brauchen nicht nach draußen fireRequest zu verwalten, SingleEmitter hat setCancellable Methode genau für das, tut die Annullierung dort, und RxJava wird sicherstellen, dass es nennen, wenn jemand die beobachtbaren entsorgen.

bei fireRequest() Methode hinzufügen und entfernen Sie die doOnDispose:

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