RxJava: पता लगाएं कि व्यवहार विषय एक दोहराया गया मान था या नहीं

Question

मैं एक एंड्रॉइड इंटरफ़ेस बना रहा हूं जो नेटवर्क से प्राप्त कुछ डेटा दिखाता है। मैं इसे नवीनतम उपलब्ध डेटा दिखाना चाहता हूं, और कभी भी खाली नहीं होना चाहिए (जब तक कि कोई डेटा अभी तक नहीं लाया गया हो) तो मैं ग्राहकों को (मेरा यूआई) नवीनतम उपलब्ध जानकारी देने के लिए एक व्यवहार विषय का उपयोग कर रहा हूं, जबकि इसे रीफ्रेश करना पृष्ठभूमि को अद्यतन करने के लिए।

यह काम करता है, लेकिन मेरे यूआई में एक और आवश्यकता के कारण, अब मुझे यह जानना है कि प्रकाशित परिणाम नेटवर्क से ताजा हो गया है या नहीं। (दूसरे शब्दों में, मुझे यह जानने की ज़रूरत है कि प्रकाशित परिणाम व्यवहारकर्ता की सहेजी गई वस्तु थी या नहीं।)

मैं इसे कैसे प्राप्त कर सकता हूं? अगर मुझे इसे कई ऑब्जर्बल्स में विभाजित करने की ज़रूरत है, तो यह ठीक है, जब तक कि मैं व्यवहारकर्ता के कैशिंग व्यवहार (अंतिम उपलब्ध परिणाम प्राप्त करना) प्राप्त करने में सक्षम हूं, जबकि यह भी बता सकता है कि परिणाम लौटाया गया है या नहीं नहीं। एक हैकी तरीका मैं ऐसा करने के बारे में सोच सकता हूं यह जांचना होगा कि प्रतिक्रिया का टाइमस्टैम्प अपेक्षाकृत जल्द ही था, लेकिन यह वास्तव में मैला होगा और मैं इसे आरएक्सजेवा के साथ करने का एक तरीका समझूंगा।

Answer 1

जैसा कि आपने इस प्रश्न में उल्लेख किया है, यह कई पर्यवेक्षकों के साथ पूरा किया जा सकता है। संक्षेप में, आपके पास दो अवलोकन हैं: "ताजा प्रतिक्रिया" देखी जा सकती है, और "कैश प्रतिक्रिया" देखी जा सकती है। अगर कुछ "मनाया जा सकता है", तो आप इसे एक अवलोकन के रूप में व्यक्त कर सकते हैं। आइए पहले नाम original और दूसरा replayed नाम दें।

यह JSBin देखें (जावास्क्रिप्ट लेकिन अवधारणाओं का सीधे जावा में अनुवाद किया जा सकता है। जहां तक ​​मुझे पता है, जावाबिन नहीं है)।

var original = Rx.Observable.interval(1000) 
    .map(function (x) { return {value: x, from: 'original'}; }) 
    .take(4) 
    .publish().refCount(); 

var replayed = original 
    .map(function (x) { return {value: x.value, from: 'replayed'}; }) 
    .replay(null, 1).refCount(); 

var merged = Rx.Observable.merge(original, replayed) 
    .replay(null, 1).refCount() 
    .distinctUntilChanged(function (obj) { return obj.value; }); 

console.log('subscribe 1st'); 
merged.subscribe(function (x) { 
    console.log('subscriber1: value ' + x.value + ', from: ' + x.from); 
}); 

setTimeout(function() { 
    console.log(' subscribe 2nd'); 
    merged.subscribe(function (x) { 
    console.log(' subscriber2: value ' + x.value + ', from: ' + x.from); 
    }); 
}, 2500); 

यहां समग्र विचार है: एक क्षेत्र from अपने मूल का संकेत के साथ घटना पर टिप्पणी करें। यदि यह original है, तो यह एक ताजा प्रतिक्रिया है। यदि यह replayed है, तो यह एक कैश प्रतिक्रिया है। अवलोकन original केवल from: 'original' उत्सर्जित करेगा और अवलोकन replayed केवल from: 'replayed' उत्सर्जित करेगा। जावा में हमें थोड़ी अधिक बॉयलरप्लेट की आवश्यकता होगी क्योंकि आपको इन एनोटेटेड घटनाओं का प्रतिनिधित्व करने के लिए कक्षा बनाने की आवश्यकता है। अन्यथा आरएक्सजेएस में एक ही ऑपरेटरों को आरएक्सजेवा में पाया जा सकता है।

मूल पर्यवेक्षण publish().refCount() है क्योंकि हम सभी पर्यवेक्षकों के साथ साझा करने के लिए इस स्ट्रीम का केवल एक उदाहरण चाहते हैं। वास्तव में आरएक्सजेएस और आरएक्स.नेट में, share()publish().refCount() के लिए उपनाम है।

रीप्लेड ऑब्जर्जेबल replay(1).refCount() है क्योंकि इसे मूल की तरह भी साझा किया जाता है, लेकिन replay(1) हमें कैशिंग व्यवहार देता है।

merged अवलोकन करने योग्य दोनों मूल और पुनः चलाए गए हैं, और यही आपको सभी ग्राहकों के सामने खुलासा करना चाहिए। चूंकि replayed तत्काल उत्सर्जित हो जाएगा जब भी original करता है, हम तत्काल अनदेखा करने के लिए ईवेंट के मान पर distinctUntilChanged का उपयोग करते हैं। कारण हम replay(1).refCount()भी विलय है क्योंकि हम सभी पर्यवेक्षकों के बीच साझा की गई स्ट्रीम का एक साझा साझा उदाहरण होने के लिए मूल और रीप्ले के विलय को भी चाहते हैं। इस उद्देश्य के लिए हमने publish().refCount() का उपयोग किया होगा, लेकिन हम replayed में रीप्ले प्रभाव को खो नहीं सकते हैं, इसलिए यह replay(1).refCount() है, publish().refCount() नहीं है।

Answer 2

एक व्यवहार संरचना में व्यवहारकर्ता वस्तु वस्तुओं की जानकारी को एक अच्छी लुकअप जैसे डिक्शनरी के साथ स्टोर करें। प्रत्येक मान एक कुंजी होगा और मूल्य पुनरावृत्ति की संख्या होगी।

वहां, जब आप एक कणिका कुंजी देखते हैं, यदि आपकी उपन्यास में पहले से ही है और इसका मान पहले से ही एक है, तो आप जानते हैं कि एक मान दोहराया गया मान है।

Answer 3

Distinct आपके मामले को कवर नहीं करता है? व्यवहार Sububject केवल सदस्यता के बाद नवीनतम तत्व दोहराता है।

Answer 4

मेरा मानना ​​है कि क्या आप चाहते हैं कुछ इस तरह है:

private final BehaviorSubject<T> fetched = BehaviorSubject.create(); 
private final Observable<FirstTime<T>> _fetched = fetched.lift(new Observable.Operator<FirstTime<T>, T>() { 
    private AtomicReference<T> last = new AtomicReference<>(); 
    @Override 
    public Subscriber<? super T> call(Subscriber<? super FirstTime<T>> child) { 
     return new Subscriber<T>(child) { 
      @Override 
      public void onCompleted() { 
       child.onCompleted(); 
      } 

      @Override 
      public void onError(Throwable e) { 
       child.onError(e); 
      } 

      @Override 
      public void onNext(T t) { 
       if (!Objects.equals(t, last.getAndSet(t))) { 
        child.onNext(FirstTime.yes(t)); 
       } else { 
        child.onNext(FirstTime.no(t)); 
       } 
      } 
     }; 
    } 
}); 

public Observable<FirstTime<T>> getObservable() { 
    return _fetched; 
} 

public static class FirstTime<T> { 
    final boolean isItTheFirstTime; 
    final T value; 

    public FirstTime(boolean isItTheFirstTime, T value) { 
     this.isItTheFirstTime = isItTheFirstTime; 
     this.value = value; 
    } 

    public boolean isItTheFirstTime() { 
     return isItTheFirstTime; 
    } 

    public T getValue() { 
     return value; 
    } 

    public static <T> FirstTime<T> yes(T value) { 
     return new FirstTime<>(true, value); 
    } 

    public static <T> FirstTime<T> no(T value) { 
     return new FirstTime<>(false, value); 
    } 
} 

आवरण वर्ग FirstTime एक बूलियन जो देखने के लिए कि प्रत्यक्ष को किसी भी ग्राहक इसे पहले देखा है इस्तेमाल किया जा सकता है।

उम्मीद है कि मदद करता है।

Answer 5

मुझे सच में यकीन नहीं है कि आप क्या हासिल करना चाहते हैं। शायद आप "नवीनतम" डेटा के लिए एक स्मार्ट स्रोत चाहते हैं और दूसरा स्रोत जो आपको बताता है कि डेटा को ताज़ा किया गया था?

BehaviorSubject<Integer> dataSubject = BehaviorSubject.create(42); // initial value, "never empty" 

    Observable<String> refreshedIndicator = dataSubject.map(data -> "Refreshed!"); 
    refreshedIndicator.subscribe(System.out::println); 

    Observable<Integer> latestActualData = dataSubject.distinctUntilChanged(); 
    latestActualData.subscribe(data -> System.out.println("Got new data: " + data)); 

    // simulation of background activity: 
    Observable.interval(1, TimeUnit.SECONDS) 
      .limit(100) 
      .toBlocking() 
      .subscribe(aLong -> dataSubject.onNext(ThreadLocalRandom.current().nextInt(2))); 

आउटपुट:

Refreshed! 
Got new data: 42 
Refreshed! 
Got new data: 0 
Refreshed! 
Refreshed! 
Refreshed! 
Got new data: 1 
Refreshed! 
Got new data: 0 
Refreshed! 
Got new data: 1