通过可观察限制重播缓冲区

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