सर्वर छोड़ने के लिए RxJava में "repeatWhen" के साथ संयुक्त "skipWhile" का उपयोग

Question

मुझे वास्तव में आरएक्सजेवा पसंद है, यह एक अद्भुत टूल है लेकिन कभी-कभी यह समझना बहुत कठिन होता है कि यह कैसे काम करता है। हम अपने एंड्रॉइड प्रोजेक्ट में आरएक्सजेवा के साथ रेट्रोफिट का उपयोग करते हैं और निम्नलिखित उपयोग-मामले हैं:

मुझे सर्वर को मतदान करने की आवश्यकता है, रिट्रीज़ के बीच कुछ देरी के साथ, सर्वर कुछ काम कर रहा है। जब सर्वर किया जाता है तो मुझे परिणाम देना होगा। इसलिए मैं सफलतापूर्वक RxJava साथ यह किया है, यहाँ कोड का टुकड़ा है: मैं "repeatWhen"

Subscription checkJobSubscription = mDataManager.checkJob(prepareTweakJob) 
     .skipWhile(new Func1<CheckJobResponse, Boolean>() { 

      @Override 
      public Boolean call(CheckJobResponse checkJobResponse) { 
       boolean shouldSkip = false; 

       if (SHOW_LOGS) Logger.v(TAG, "checkJob, skipWhile, jobStatus " + checkJobResponse.getJobStatus()); 

       switch (checkJobResponse.getJobStatus()){ 
        case CheckJobResponse.PROCESSING: 
         shouldSkip = true; 
         break; 
        case CheckJobResponse.DONE: 
        case CheckJobResponse.ERROR: 
         shouldSkip = false; 
         break; 
       } 
       if (SHOW_LOGS) Logger.v(TAG, "checkJob, skipWhile, shouldSkip " + shouldSkip); 

       return shouldSkip; 
      } 
     }) 
     .repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() { 
      @Override 
      public Observable<?> call(Observable<? extends Void> observable) { 
       if (SHOW_LOGS) Logger.v(TAG, "checkJob, repeatWhen " + observable); 
       return observable.delay(1, TimeUnit.SECONDS); 
      } 
     }).subscribe(new Subscriber<CheckJobResponse>(){ 
      @Override 
      public void onNext(CheckJobResponse response) { 
       if (SHOW_LOGS) Logger.v(TAG, "checkJob, onSuccess, response " + response); 

      } 

      @Override 
      public void onError(BaseError error) { 
       if (SHOW_LOGS) Logger.v(TAG, "checkJob, onError, canEditTimeline, error " + error); 
       Toast.makeText(ChoseEditOptionActivity.this, R.string.NETWORK__no_internet_message, Toast.LENGTH_LONG).show(); 

      } 

      @Override 
      public void onCompleted() { 
       if (SHOW_LOGS) Logger.v(TAG, "onCompleted"); 
      } 
     }); 

के साथ "skipWhile" का इस्तेमाल किया कोड ठीक काम करता है: मैं

जब सर्वर ने जवाब दिया कि काम संसाधित कर रहा है "skipWhile" श्रृंखला से "सत्य" लौटें, मूल पर्यवेक्षक 1 सेकंड के लिए प्रतीक्षा करता है और फिर से http अनुरोध करता है। यह प्रक्रिया दोहराई जाती है जब तक कि मैं "skipWhile" श्रृंखला से "झूठी" वापस नहीं आती।

यहाँ कुछ चीजें है मुझे समझ नहीं आता:

1. मैं "skipWhile" के दस्तावेज में देखा है कि यह मूल प्रत्यक्ष से कुछ भी (onError, onNext, onComplete) का उत्सर्जन नहीं होगा जब तक मैं वापस जाने के इसकी "कॉल" विधि से "झूठा"। तो अगर यह कुछ भी नहीं निकलता है तो "दोहराना कब" पर्यवेक्षक इसे नौकरी कर रहा है? यह एक सेकंड के लिए इंतजार कर रहा है और अनुरोध फिर से चलाता है। इसे किसने लॉन्च किया?

2. दूसरा सवाल यह है: क्यों "दोहराना" हमेशा से नहीं चल रहा है, मेरा मतलब है कि जब मैं "skipWhile" से "झूठी" वापस लौटता हूं तो क्यों दोहराना बंद कर देता है? अगर मैं "झूठी" लौटाता हूं तो मैं अपने सब्सक्राइबर में सफलतापूर्वक आगे बढ़ता हूं।

3. "दोहराना समय" के प्रलेखन में यह कहता है कि अंत में मुझे अपने सब्सक्राइबर में "पूर्ण" पर कॉल करने के लिए कॉल किया जाता है लेकिन "ऑन कॉम्प्ले" कभी नहीं कहा जाता है।

4. अगर मैं "skipWhile" और "repeatWhen" चेनिंग के क्रम को बदलता हूं तो इससे कोई फर्क नहीं पड़ता। ऐसा क्यों है ?

मैं समझता हूं कि आरएक्सजेवा ओपनसोर्स है और मैं सिर्फ कोड पढ़ सकता हूं, लेकिन जैसा कि मैंने कहा - यह समझना वाकई मुश्किल है।

धन्यवाद।

Answer 1

मैंने पहले repeatWhen के साथ काम नहीं किया है, लेकिन इस सवाल ने मुझे उत्सुक बना दिया, इसलिए मैंने कुछ शोध किया।

skipWhile फेंकना onError और onCompleted, भले ही वह कभी नहीं तो पहले true देता है। इस प्रकार, repeatWhen हर बार checkJob() उत्सर्जित किया जा रहा है onCompleted। वह प्रश्न # 1 का जवाब देता है।

शेष प्रश्न झूठी धारणाओं पर आधारित हैं। आपकी सदस्यता हमेशा के लिए चल रही है क्योंकि आपका repeatWhen कभी समाप्त नहीं होता है। ऐसा इसलिए है क्योंकि repeatWhen आपके एहसास से अधिक जटिल जानवर है। इसमें Observable स्रोत से onCompleted प्राप्त होने पर null उत्सर्जित करता है। यदि आप इसे लेते हैं और onCompleted लौटाते हैं तो यह समाप्त होता है, अन्यथा यदि आप कुछ भी छोड़ देते हैं तो यह पुनः प्रयास करता है। चूंकि delay सिर्फ उत्सर्जन लेता है और इसे देरी करता है, यह हमेशा null उत्सर्जित करता है।इस प्रकार, यह लगातार सदस्यता लेता है।

# 2 का उत्तर, तो, यह हमेशा के लिए चल रहा है; सदस्यता रद्द करने के लिए आप शायद इस कोड के बाहर कुछ और कर रहे हैं। # 3 के लिए, आपको कभी भी onCompleted नहीं मिलता क्योंकि यह कभी पूरा नहीं होता है। # 4 के लिए, आदेश कोई फर्क नहीं पड़ता क्योंकि आप अनिश्चित काल तक दोहरा रहे हैं।

सवाल यह है कि, आप सही व्यवहार कैसे प्राप्त करते हैं? skipWhile के बजाय takeUntil का उपयोग करना उतना आसान है। इस तरह, आप को तक दोहराते रहें, आपको वह परिणाम मिल जाएगा जो आप चाहते हैं, इस प्रकार स्ट्रीम को समाप्त कर दें जब आप इसे समाप्त करना चाहते हैं।

Observable<Boolean> source = ...; // Something that eventually emits true 

source 
    .repeatWhen(completed -> completed.delay(1, TimeUnit.SECONDS)) 
    .takeUntil(result -> result) 
    .filter(result -> result) 
    .subscribe(
     res -> System.out.println("onNext(" + res + ")"), 
     err -> System.out.println("onError()"), 
     () -> System.out.println("onCompleted()") 
    ); 

इस उदाहरण में, source बूलियन्स उत्सर्जित करती है:

यहाँ एक कोड नमूना है। जब तक स्रोत true उत्सर्जित नहीं करता तब तक मैं हर 1 सेकंड दोहराता हूं। मैं resulttrue तक लेता रहता हूं। और मैं false पर सभी सूचनाएं फ़िल्टर करता हूं, इसलिए ग्राहक true तक उन्हें प्राप्त नहीं करते हैं।