LatestFromで観測可能

Question

I kotlinの拡張関数として "FilterByLatestFrom"という擬似演算子を実装しました。

私は、この演算子を使用して次のコードを書いた：

fun testFilterByLatestFromOperator(){ 
    val observableA : Observable<Int> = Observable.fromArray(1,2,3,4,5,6,7,8,9,10) 
    val observableC : PublishSubject<Int> = PublishSubject.create() 
    val observableB : Observable<Int> = Observable.just(2).mergeWith(observableC) 

    observableB.subscribe { println("observableB onNext: $it") } 

    observableA 
      .subscribe({ println("Original : $it")}) 

    observableA.filterByLatestFrom(observableB, BiFunction { aVal, bVal -> aVal%bVal==0 }) 
      .subscribe({ println("Result A : $it") }) 

    observableC.onNext(3) 

    observableA.filterByLatestFrom(observableB, BiFunction { aVal, bVal -> aVal%bVal==0 }) 
      .subscribe({ println("Result AC : $it") }) 
} 

出力は次のとおりです。

observableB onNext: 2 
Original : 1 
Original : 2 
Original : 3 
Original : 4 
Original : 5 
Original : 6 
Original : 7 
Original : 8 
Original : 9 
Original : 10 
Result A : 2 
Result A : 4 
Result A : 6 
Result A : 8 
Result A : 10 
observableB onNext: 3 
Result AC : 2 
Result AC : 4 
Result AC : 6 
Result AC : 8 
Result AC : 10 

私はフィルタ演算子が観察Bの最後の値に従ってobsAフィルタします それはのために働くしたいです最初のブロックですが、新しい値でOn-nextを追加しても、結果は変わりません（元の観測値と同じ最終値を使用します）。

これは（それは（）を構成しても、Javaから使用するデザインはFilterByLatestFromのIMPLたです：

class FilterByLatestFrom<T,U>(private val observable: Observable<T>, private val biFunction: BiFunction<U, T, Boolean>){ 
fun filter() : ObservableTransformer<U,U> = ObservableTransformer { 
    it 
      .withLatestFrom(
        observable, 
        BiFunction<U,T,Pair<U,Boolean>> { 
         u, t -> Pair(u,biFunction.apply(u,t)) 
        }) 
      .filter { it.second } 
      .map { it.first } 
    } 
} 
fun <T,U> Observable<U>.filterByLatestFrom(observable: Observable<T>, biFunction: BiFunction<U, T, Boolean>) : Observable<U> = 
     this.compose(FilterByLatestFrom(observable,biFunction).filter()) 

私は何をしないのです

EDIT：？私は、問題を見つけたと思います： PublishSubjectはobsCがobsB後に発することを約束するconacatしなければならない代わりにBehaviorSubjectこと。そして機能をマージする必要があります。

Answer 1

あなたの擬似オペレータfilterByLatestFrom問題がテスト内にある、だけで結構です、PublishSubjectが放出されます最後のサブスクリプション（ '結果AC'）でobservableBはすでに2を放出します。observableCはすでに3を放出しており、observableB（mergeを使用）には再生されません。

最後のサブスクリプション（最終行）の後にobservableC.onNext(3)を移動するだけで、予期した動作が表示されます。

EDIT：あなたは同じ問題を解決しましたように もPublishSubjectへの変更（対象者が新しいサブスクリプションへの最後の値を再生します）