通過可觀察限制重播緩衝區

Question

我有一個帶有實時數據的流，以及基本上分隔屬於一起的實時數據部分的流。現在當有人訂閱實時數據流時，我想重播他們的實時數據。但是我不想記住所有的實時數據，只有自上一次其他流發出值以來的部分。

There is an issue這將解決我的問題，因爲有一個重播操作符正是我想要的（或至少我認爲）。

目前如何輕鬆完成此操作？有沒有比以下更好的方法？

private class ReplayWithLimitObservable<TItem, TDelimiter> : IConnectableObservable<TItem> 
{ 
    private readonly List<TItem> cached = new List<TItem>(); 
    private readonly IObservable<TDelimiter> delimitersObservable; 
    private readonly IObservable<TItem> itemsObservable; 
    public ReplayWithLimitObservable(IObservable<TItem> itemsObservable, IObservable<TDelimiter> delimitersObservable) 
    { 
     this.itemsObservable = itemsObservable; 
     this.delimitersObservable = delimitersObservable; 
    } 

    public IDisposable Subscribe(IObserver<TItem> observer) 
    { 
     lock (cached) 
     { 
      cached.ForEach(observer.OnNext); 
     } 

     return itemsObservable.Subscribe(observer); 
    } 

    public IDisposable Connect() 
    { 
     var delimiters = delimitersObservable.Subscribe(
      p => 
       { 
        lock (cached) 
        { 
         cached.Clear(); 
        } 
       }); 
     var items = itemsObservable.Subscribe(
      p => 
       { 
        lock (cached) 
        { 
         cached.Add(p); 
        } 
       }); 
     return Disposable.Create(
      () => 
       { 
        items.Dispose(); 
        delimiters.Dispose(); 
        lock (cached) 
        { 
         cached.Clear(); 
        } 
      }); 
} 

public static IConnectableObservable<TItem> ReplayWithLimit<TItem, TDelimiter>(IObservable<TItem> items, IObservable<TDelimiter> delimiters) 
{ 
    return new ReplayWithLimitObservable<TItem, TDelimiter>(items, delimiters); 
} 

Answer 1

這是做你想做的嗎？它使所有的鎖定和競爭條件到Rx利弊:)

private class ReplayWithLimitObservable<T, TDelimiter> : IConnectableObservable<T> 
{ 
    private IConnectableObservable<IObservable<T>> _source; 

    public ReplayWithLimitObservable(IObservable<T> source, IObservable<TDelimiter> delimiter) 
    { 
    _source = source 
     .Window(delimiter) // new replay window on delimiter 
     .Select<IObservable<T>,IObservable<T>>(window => 
     { 
     var replayWindow = window.Replay(); 

     // immediately connect and start memorizing values 
     replayWindow.Connect(); 

     return replayWindow; 
     }) 
     .Replay(1); // remember the latest window 
    } 

    IDisposable Connect() 
    { 
    return _source.Connect(); 
    } 

    IDisposable Subscribe(IObserver<T> observer) 
    { 
    return _source 
     .Concat() 
     .Subscribe(observer); 
    } 
} 

public static IConnectableObservable<TItem> ReplayWithLimit<TItem, TDelimiter>(IObservable<TItem> items, IObservable<TDelimiter> delimiters) 
{ 
    return new ReplayWithLimitObservable<TItem, TDelimiter>(items, delimiters); 
}