CombineLatest, लेकिन बाएं

Question

के लिए केवल पुश करें, मुझे CombineLatest का संस्करण लागू करने की आवश्यकता है (मैं इसे WithLatest यहां कॉल करूंगा) जो बाईं ओर प्रत्येक आइटम और दाईं ओर नवीनतम आइटम के लिए चयनकर्ता को कॉल करता है। इसे केवल सही परिवर्तन पर वस्तुओं के लिए धक्का नहीं देना चाहिए।

मुझे लगता है कि इस Observable.Create बनाया गया है या मौजूदा एक्सटेंशन का एक संयोजन विशेष रूप से महत्वपूर्ण नहीं है कि; मैं इसे किसी भी तरह से "बॉक्सिंग" एक्सटेंशन विधि बनाउंगा।

उदाहरण

var left = new Subject<int>(); 
var right = new Subject<int>(); 

left.WithLatest(right, (l,r) => l + " " + r).Dump(); 

left.OnNext(1); // <1> 
left.OnNext(2); // <2> 
right.OnNext(1); // <3> 
right.OnNext(2); // <4> 
left.OnNext(3); // <5> 

उपज चाहिए

2 1 
3 2 

संपादित: मेरे उदाहरण के तर्क जाता है:

1. वाम के साथ 1. राइट खाली है आबादी वाले हो जाता है, कोई मूल्य धक्का नहीं दिया।
2. बाएं 2 के साथ अपडेट हो जाता है (यह पिछले मान को भूल जाता है)। अभी भी खाली है, इसलिए कुछ भी धक्का नहीं दिया जाता है।
3. दाएं 1 के साथ आबादी बन जाती है, इसलिए बाएं = 2 (नवीनतम मान), दायां = 1 धक्का दिया जाता है। इस बिंदु तक, वहाँ कोई अंतर नहीं WithLatest और
4. के बीच CombineLatest सही अद्यतन किया जाता है है - कुछ भी नहीं धकेल दिया जाता है। यह वही विभिन्न
5. वाम 3 के साथ अद्यतन किया जाता है है, तो वाम = 3, सही = 2 (नवीनतम मूल्य) धकेल दिया जाता है।

---

यह सुझाव हो गया है कि मैं कोशिश:

var lr = right.ObserveOn(Scheduler.TaskPool).Latest(); 
left.Select(l => l + " " + lr.First()).Dump(); 

लेकिन अपने परीक्षण के लिए मौजूदा धागे पर इस ब्लॉक।

Answer 1

यहाँ बनाएं का उपयोग कर hacky तरीका है - वास्तव में यह निर्माण नहीं किया, यह वास्तव में काम करता है तो विदेश मंत्रालय culpa नहीं :)

public static IObservable<TRet> WithLatest<TLeft, TRight, TRet>(
     this IObservable<TLeft> lhs, 
     IObservable<TRight> rhs, 
     Func<TLeft, TRight, TRet> sel) 
{ 
    return Observable.Create<TRet>(subj => { 
     bool rhsSet = false; 
     bool deaded = false; 
     var latestRhs = default(TRight); 

     Action onDeaded = null; 

     var rhsDisp = rhs.Subscribe(
      x => { latestRhs = x; rhsSet = true; }, 
      ex => { subj.OnError(ex); onDeaded(); }); 

     var lhsDisp = lhs 
      .Where(_ => deaded == false && rhsSet == true) 
      .Subscribe(
       x => subj.OnNext(sel(x, latestRhs)), 
       ex => { subj.OnError(ex); onDeaded(); }, 
       () => { subj.OnCompleted(); onDeaded(); }); 

     onDeaded =() => { 
      deaded = true; 
      if (lhsDisp != null) { 
       lhsDisp.Dispose(); 
       lhsDisp = null; 
      } 
      if (rhsDisp != null) { 
       rhsDisp.Dispose(); 
       rhsDisp = null; 
      } 
     }; 

     return onDeaded; 
    }); 
} 

Answer 2

मैं भी जो "केवल के लिए धक्का एक CombineLatest के लिए एक ही जरूरत थी छोडा"।

मैं, समाधान एक Observable.Sample की "अधिभार" हो पाता है क्योंकि है कि क्या विधि करता है:
यह नमूने एक source (दाएं) एक sampler (बाएं) के साथ, एक resultSelector (CombineLatest में की तरह प्रदान करने के अतिरिक्त क्षमता के साथ)।

public static IObservable<TResult> Sample<TSource, TSample, TResult>(
    this IObservable<TSource> source, 
    IObservable<TSample> sampler, 
    Func<TSource, TSample, TResult> resultSelector) 
{ 
    var multiSampler = sampler.Publish().RefCount(); 
    return source.CombineLatest(multiSampler, resultSelector).Sample(multiSampler); 
} 

Answer 3

समाधान पोस्ट लेखक द्वारा उठाया मुझे लगता है कि वहाँ एक और भी सरल समाधान DistinctUntilChanged उपयोग के आधार पर:

public static IObservable<TResult> CombineLatestOnLeft<TLeft, TRight, TResult>(this IObservable<TLeft> leftSource, IObservable<TRight> rightSource, Func<TLeft, TRight, TResult> selector) { 
     return leftSource 
      .Select<TLeft, Tuple<TLeft, int>>(Tuple.Create<TLeft, int>) 
      .CombineLatest(rightSource, 
       (l, r) => new { Index = l.Item2, Left = l.Item1, Right = r }) 
      .DistinctUntilChanged(x => x.Index) 
      .Select(x => selector(x.Left, x.Right)); 
    } 

या यहाँ तक कि

public static IObservable<TResult> CombineLatestOnLeft<TLeft, TRight, TResult>(this IObservable<TLeft> leftSource, IObservable<TRight> rightSource, Func<TLeft, TRight, TResult> selector) { 
     return leftSource 
      .CombineLatest(rightSource, 
       (l, r) => new { Left = l, Right = r }) 
      .DistinctUntilChanged(x => x.Left) 
      .Select(x => selector(x.Left, x.Right)); 
    } 

अगर आप केवल के विशिष्ट मानों के बारे में परवाह leftSource

Answer 4

आप मौजूदा ऑपरेटो का उपयोग कर ऐसा कर सकते हैं रु।

Func<int, int, string> selector = (l, r) => l + " " + r; 

var query = right.Publish(rs => left.Zip(rs.MostRecent(0), selector).SkipUntil(rs)); 

• Publish सुनिश्चित करता है कि हम केवल कभी एक बार right की सदस्यता और rs करने के लिए सभी ग्राहकों के बीच में सदस्यता को साझा करें।

• MostRecent एक IEnumerable<T> कि हमेशा स्रोत नमूदार से सबसे हाल ही में उत्सर्जित मूल्य पैदावार में एक IObservable<T> बदल जाता है।

• ZipIObservable<T> और IEnumerable<U> के बीच प्रत्येक बार जब मूल्य देखने योग्य हो जाता है तो मान मानता है।

• SkipUntil जोड़े (l, r) जो right से पहले होता है, कभी भी मूल्य को छोड़ देता है।

Answer 5

मैंने आज परियोजना के लिए एक आरएक्स ऑपरेटर बनाया है जो ऐसा करता है।

यहाँ मेरी समाधान है:

public static IObservable<Tuple<TSource, TTarget>> JoinLeftSoft<TSource, TTarget>(
     this IObservable<TSource> source, IObservable<TTarget> right) 
    { 
     return source 
      .Select(x => new Tuple<object, TSource>(new object(), x)) 
      .CombineLatest(right, (l, r) => new Tuple<object, TSource, TTarget>(l.Item1, l.Item2, r)) 
      .DistinctUntilChanged(t => t.Item1) 
      .Select(t => new Tuple<TSource, TTarget>(t.Item2, t.Item3)); 
    } 

Answer 6

नवीनतम पर System.Reactive, हम WithLatestFrom विस्तार विधि का उपयोग कर सकते हैं।

left.WithLatestFrom(right, (l, r) => l + " " + r).Dump(); 

परिणाम सही ढंग से नीचे होगा।

3 2