CombineLatest, nhưng chỉ đẩy cho bên trái

Question

Tôi cần triển khai phiên bản CombineLatest (Tôi sẽ gọi nó là WithLatest tại đây) gọi bộ chọn cho mọi mục ở bên trái và mục mới nhất ở bên phải. Nó không nên đẩy cho các hạng mục bên phải chỉ thay đổi.

Tôi nghĩ rằng điều này được xây dựng Observable.Create hoặc kết hợp các tiện ích mở rộng hiện tại không đặc biệt quan trọng; Tôi sẽ làm cho điều này là một phương pháp mở rộng "đóng hộp" một trong hai cách.

Ví dụ

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> 

nên năng suất

2 1 
3 2 

Sửa: Logic của ví dụ của tôi đi:

1. Left trở nên dân cư với 1. Phải là trống rỗng, không có giá trị nào được đẩy.
2. Trái được cập nhật với 2 (nó quên giá trị trước đó). Ngay bây giờ vẫn còn trống rỗng, vì vậy không có gì được đẩy.
3. Phải trở thành dân cư với 1, vì vậy Trái = 2 (giá trị mới nhất), Phải = 1 được đẩy. Đến thời điểm này, không có sự khác biệt giữa WithLatest và CombineLatest
4. Quyền được cập nhật - không có gì được đẩy. Đây là những gì khác nhau
5. Còn lại được cập nhật với 3, vì vậy Trái = 3, Phải = 2 (giá trị mới nhất) được đẩy.

---

Nó được đề nghị tôi thử:

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

nhưng khối này trên thread hiện hành đối với thử nghiệm của tôi.

Answer 1

Dưới đây là cách sử dụng hacky Tạo - không thực sự xây dựng nó, mea culpa nếu nó không thực sự làm việc :)

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

Tôi cũng có nhu cầu tương tự cho một CombineLatest mà "đẩy chỉ cho Bên trái".

tôi đã thực hiện các giải pháp một "quá tải" của Observable.Sample, bởi vì đó là những gì phương pháp thực hiện:
Nó mẫu một source (bên phải) với một sampler (trái), với khả năng bổ sung cung cấp một resultSelector (như trong 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

Dựa trên các giải pháp chọn bởi những bài tác giả Tôi nghĩ rằng có một giải pháp đơn giản hơn sử dụng 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)); 
    } 

hoặc thậm chí

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)); 
    } 

nếu bạn chỉ quan tâm đến giá trị khác biệt của leftSource

Answer 4

Bạn có thể thực hiện việc này bằng cách sử dụng toán tử hiện có rs.

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

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

• Publish đảm bảo chúng tôi chỉ bao giờ đăng ký right một lần và chia sẻ các thuê bao trong số tất cả các thuê bao để rs.

• MostRecent biến IObservable<T> thành IEnumerable<T> luôn mang lại giá trị được phát gần đây nhất từ ​​nguồn có thể quan sát được.

• Zip giữa IObservable<T> và IEnumerable<U> phát ra giá trị mỗi khi có thể quan sát phát ra giá trị.

• SkipUntil bỏ qua các cặp (l, r) xảy ra trước right bao giờ phát ra giá trị.

Answer 5

Tôi đã tạo một toán tử RX cho dự án ngày nay thực hiện điều này.

Đây là giải pháp của tôi:

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

On mới nhất System.Reactive, chúng ta có thể sử dụng WithLatestFrom phương pháp khuyến nông.

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

Kết quả sẽ dưới chính xác.

3 2