.flatMap के बिना प्रवाह को नियंत्रित करने के लिए, जो एक धारावाहिक धारा को तोड़ता है जैसे ऑपरेटरों को अलग-अलग स्ट्रीम पर रोकता है, पूरे स्ट्रीम

Question

पर काम करने से बदलकर मैं State के विभिन्न कार्यान्वयन के लिए तर्क की एक अलग देखने योग्य श्रृंखला को संभालना चाहता हूं। इसे आसानी से एक सीलबंद वर्ग/बीजगणितीय डेटा प्रकार/संघ + .flatMap() के साथ हासिल किया जा सकता है, लेकिन यह धारा को तोड़ता है, जहां .distinctUntilChanged() जैसे ऑपरेटर केवल फ़ंक्शन के भीतर काम करते हैं, पूरे स्ट्रीम पर ही नहीं।

sealed class State { 
    object Loading : State() 
    data class Loaded(val value: Int) : State() 
} 

@Test fun distinctTest() { 
    val relay = PublishRelay.create<State>() 
    relay.flatMap { 
     fun handle(state: State): Observable<*> = when (state) { 
      State.Loading -> Observable.just(state) 
        .distinctUntilChanged() 
        .doOnNext { println("loading") } 

      is State.Loaded -> Observable.just(state) 
        .distinctUntilChanged() 
        .doOnNext { println(it.value) } 
     } 
     handle(it) 
    } 
      .subscribe() 

    relay.accept(State.Loading) 
    relay.accept(State.Loaded(1)) 
    relay.accept(State.Loaded(2)) 
    relay.accept(State.Loaded(3)) 
    relay.accept(State.Loaded(3)) 
    //desired: loading, 1, 2, 3 
    //actual: loading, 1, 2, 3, 3 
} 

नोट, यह एक सरलीकृत उदाहरण है। जबकि मैं बस यहां प्रिंट कर रहा हूं, मैं वास्तव में कार्यान्वयन प्रकार State

पर आधारित विभिन्न कार्यों (यूआई को अलग-अलग प्रस्तुत करना) करना चाहता हूं, यह किसी विषय/रिले के साथ पूरा किया जा सकता है, लेकिन यह डिस्कनेक्ट हो जाएगा, उत्परिवर्तनीय धारा, जिसे मैं भी टालना चाहूंगा।

Answer 1

क्या आप Observable को कई अवलोकनों में विभाजित कर सकते हैं जिनमें से प्रत्येक एक प्रकार की घटनाएं प्राप्त कर रहा है? फिर आप इन अवलोकनों पर फिर से एक साथ विलय करने से पहले कुछ संचालन कर सकते हैं।

मैं अभी इस परीक्षण का परीक्षण नहीं कर पा रहा हूं, इसलिए इसे कुछ बदलावों की आवश्यकता हो सकती है। वैसे भी, मुझे आशा है कि आपको यहां विचार मिलेगा:

@Test fun distinctTest() { 
    val relay = PublishRelay.create<State>() 

    val loadingObs = relay.filter { it is State.Loading } 
          .distinctUntilChanged() 
          .doOnNext { println("loading") } 

    val loadedObs = relay.filter { it is State.Loaded } 
         .distinctUntilChanged() 
         .doOnNext { println(it.value) } 

    val merged = loadingObs.mergeWith(loadedObs) 

    merged.subscribe() 

    relay.accept(State.Loading) 
    relay.accept(State.Loaded(1)) 
    relay.accept(State.Loaded(2)) 
    relay.accept(State.Loaded(3)) 
    relay.accept(State.Loaded(3)) 
    // Hopefully prints this: loading, 1, 2, 3 
}