通过 observable 限制重放缓冲区 [英] Limit replay buffer by observable

问题描述

我有一个包含实时数据的流，以及一个基本上界定属于一起的实时数据部分的流.现在，当有人订阅实时数据流时，我想重播他们的实时数据.但是，我不想记住所有实时数据，只记住自上次另一个流发出值以来的部分.

I have a stream with live data, and a stream which basically delimits parts of the live data that belong together. Now when someone subscribes to the live data stream, I would like to replay them the live data. However I don't want to remember all the live data, only the part since the last time the other stream emitted a value.

有一个问题可以解决我的问题，因为有一个重放操作符正是我想要的(或至少我认为).

There is an issue which would solve my problem, since there is a replay operator which does exactly what I want (or at least I think).

目前有什么方法可以轻松做到这一点?还有比下面更好的方法吗?

What is currently the way to do this easily? Is there a better way than something like the following?

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

推荐答案

这符合您的要求吗?它的优点是将所有锁定和竞争条件留给 Rx 专家:)

Does this do what you want? It has the advantage of leaving all of the locking and race conditions to the Rx pros :)

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

这篇关于通过 observable 限制重放缓冲区的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！