缓存将更新和新值作为“distinctlatest”;以及订阅后的完整缓存内容

public static class RxExtensions
{
    /// <summary>
    /// A cache that keeps distinct elements where the elements are replaced by the latest. Upon subscription the subscriber should receive the full cache contents.
    /// </summary>
    /// <typeparam name="T">The type of the result</typeparam>
    /// <typeparam name="TKey">The type of the selector key for distinct results.</typeparam>
    /// <param name="newElements">The sequence of new elements.</param>
    /// <param name="seedElements">The elements when the cache is started.</param>
    /// <param name="replacementSelector">The replacement to select distinct elements in the cache.</param>
    /// <returns>The cache contents upon first call and changes thereafter.</returns>
    public static IObservable<T> Cache<T, TKey>(this IObservable<T> newElements, IEnumerable<T> seedElements, Func<T, TKey> replacementSelector)
    {
        var replaySubject = new ReplaySubject<T>();
        seedElements.ToObservable().Concat(newElements).Subscribe(replaySubject);
        return replaySubject.Distinct(replacementSelector);
    }
}

newElements.Subscribe(replaySubject);
return replaySubject.Concat(seedElements.ToObservable()).Distinct(replacementSelector);

public void CacheTests()
{
    var seedElements = new List<Event>(new[]
    {
        new Event { Id = 0, Batch = 1 },
        new Event { Id = 1, Batch = 1 },
        new Event { Id = 2, Batch = 1 }
    });
    var testScheduler = new TestScheduler();
    var observer = testScheduler.CreateObserver<Event>();
    var batchTicks = TimeSpan.FromSeconds(10);
    var xs = testScheduler.CreateHotObservable
    (
        ReactiveTest.OnNext(batchTicks.Ticks, new Event { Id = 0, Batch = 2 }),
        ReactiveTest.OnNext(batchTicks.Ticks, new Event { Id = 1, Batch = 2 }),
        ReactiveTest.OnNext(batchTicks.Ticks, new Event { Id = 2, Batch = 2 }),
        ReactiveTest.OnNext(batchTicks.Ticks, new Event { Id = 3, Batch = 2 }),
        ReactiveTest.OnNext(batchTicks.Ticks, new Event { Id = 4, Batch = 2 }),
        ReactiveTest.OnNext(batchTicks.Ticks + 10, new Event { Id = 0, Batch = 3 }),
        ReactiveTest.OnNext(batchTicks.Ticks + 10, new Event { Id = 1, Batch = 3 })
    );
    var subs = xs.Cache(seedElements, i => i.Id).Subscribe(observer);
    var seedElementsAndNoMore = observer.Messages.ToArray();
    Assert.IsTrue(observer.Messages.Count == 3);
    testScheduler.Start();
    var seedAndReplacedElements = observer.Messages.ToArray();
    //OK, a bad assert, we should create expected timings and want to check
    //also the actual batch numbers, but to get things going...
    //There should be Events with IDs { 1, 2, 3, 4, 5 } all having a batch number
    //of either 2 or 3. Also, a total of 7 (not 10) events
    //should've been observed.
    Assert.IsTrue(observer.Messages.Count == 7);
    for(int i = 0; i < seedAndReplacedElements.Length; ++i)
    {                
        Assert.IsTrue(seedAndReplacedElements[i].Value.Value.Batch > 1)             
    }
}

public static IObservable<T> Cache<T, TKey>(this IObservable<T> newElements, IEnumerable<T> seedElements, Func<T, TKey> replacementSelector)
{
    var replaySubject = new ReplaySubject<T>();
    newElements.StartWith(seedElements).Distinct(replacementSelector).Subscribe(replaySubject);
    return replaySubject;           
}

好的，我想我有你想要的东西。我主要是从下面这句话来确定你的需求的:

当订阅者订阅此缓存时，它首先获取缓存中保存的所有值，然后在它们进入时更新

我相信这是希望在单一订阅之外有一个生命周期(即它应该启动，订阅者可以随意来和去)，因此使它成为一个IConnectableObservable(这在你的代码中是隐含的，但没有正确的作用域)。

我还重构了您的测试，以显示多个订阅者(根据@Shlomo的评论)，如下所示:

[Fact]
public void ReplayAllElements()
{
    var seedElements = new List<Event>(new[]
    {
        new Event { Id = 0, Batch = 1 },
        new Event { Id = 1, Batch = 1 },
        new Event { Id = 2, Batch = 1 }
    });
    var testScheduler = new TestScheduler();
    var xs = testScheduler.CreateHotObservable
    (
        ReactiveTest.OnNext(1, new Event { Id = 0, Batch = 2 }),
        ReactiveTest.OnNext(2, new Event { Id = 1, Batch = 2 }),
        ReactiveTest.OnNext(3, new Event { Id = 2, Batch = 2 }),
        ReactiveTest.OnNext(4, new Event { Id = 3, Batch = 2 }),
        ReactiveTest.OnNext(5, new Event { Id = 4, Batch = 2 }),    
        ReactiveTest.OnNext(6, new Event { Id = 0, Batch = 3 }),
        ReactiveTest.OnNext(7, new Event { Id = 1, Batch = 3 })
    );
    IConnectableObservable<Event> cached = xs.Cache(seedElements, i => i.Id);
    var observerA = testScheduler.CreateObserver<Event>();
    cached.Subscribe(observerA);
    cached.Connect();
    testScheduler.AdvanceTo(4);
    var observerB = testScheduler.CreateObserver<Event>();
    cached.Subscribe(observerB);
    testScheduler.AdvanceTo(7);
    var expectedA = new[]
    {
        ReactiveTest.OnNext<Event>(0, @event => @event.Id == 0 && @event.Batch == 1 ),
        ReactiveTest.OnNext<Event>(0, @event => @event.Id == 1 && @event.Batch == 1 ),
        ReactiveTest.OnNext<Event>(0, @event => @event.Id == 2 && @event.Batch == 1 ),
        ReactiveTest.OnNext<Event>(1, @event => @event.Id == 0 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(2, @event => @event.Id == 1 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(3, @event => @event.Id == 2 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(4, @event => @event.Id == 3 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(5, @event => @event.Id == 4 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(6, @event => @event.Id == 0 && @event.Batch == 3 ),
        ReactiveTest.OnNext<Event>(7, @event => @event.Id == 1 && @event.Batch == 3 )
    };
    observerA.Messages.AssertEqual(expectedA);
    var expectedB = new[]
    {
        ReactiveTest.OnNext<Event>(5, @event => @event.Id == 0 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(5, @event => @event.Id == 1 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(5, @event => @event.Id == 2 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(5, @event => @event.Id == 3 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(5, @event => @event.Id == 4 && @event.Batch == 2 ),
        ReactiveTest.OnNext<Event>(6, @event => @event.Id == 0 && @event.Batch == 3 ),
        ReactiveTest.OnNext<Event>(7, @event => @event.Id == 1 && @event.Batch == 3 )
    };
    observerB.Messages.AssertEqual(expectedB);
}

如您所见，observerA获取所有种子值和更新，而observerB仅获取每个键的最新值，然后进行进一步更新。

执行此操作的代码如下:

public static class RxExtensions
{
    /// <summary>
    /// A cache that keeps distinct elements where the elements are replaced by the latest.
    /// </summary>
    /// <typeparam name="T">The type of the result</typeparam>
    /// <typeparam name="TKey">The type of the selector key for distinct results.</typeparam>
    /// <param name="newElements">The sequence of new elements.</param>
    /// <param name="seedElements">The elements when the cache is started.</param>
    /// <param name="keySelector">The replacement to select distinct elements in the cache.</param>
    /// <returns>The cache contents upon first call and changes thereafter.</returns>
    public static IConnectableObservable<T> Cache<T, TKey>(this IObservable<T> newElements, IEnumerable<T> seedElements, Func<T, TKey> keySelector)
    {
        return new Cache<TKey, T>(newElements, seedElements, keySelector);
    }
}
public class Cache<TKey, T> : IConnectableObservable<T>
{
    private class State
    {
        public ImmutableDictionary<TKey, T> Cache { get; set; }
        public T Value { get; set; }
    }
    private readonly IConnectableObservable<State> _source;
    private readonly IObservable<T> _observable;
    public Cache(IObservable<T> newElements, IEnumerable<T> seedElements, Func<T, TKey> keySelector)
    {
        var agg = new State { Cache = seedElements.ToImmutableDictionary(keySelector), Value = default(T) };
        _source = newElements
            // Use the Scan operator to update the dictionary of values based on key and use the anonymous tuple to pass this and the current item to the next operator
            .Scan(agg, (tuple, item) => new State { Cache = tuple.Cache.SetItem(keySelector(item), item), Value = item })
            // Ensure we always have at least one item
            .StartWith(agg)
            // Share this single subscription to the above with all subscribers
            .Publish();
        _observable = _source.Publish(source =>
                // ... concatting ...
                Observable.Concat(
                    // ... getting a single collection of values from the cache and flattening it to a series of values ...
                    source.Select(tuple => tuple.Cache.Values).Take(1).SelectMany(values => values),
                    // ... and the returning the values as they're emitted from the source
                    source.Select(tuple => tuple.Value)
                )
            );
    }
    public IDisposable Connect()
    {
        return _source.Connect();
    }
    public IDisposable Subscribe(IObserver<T> observer)
    {
        return _observable.Subscribe(observer);
    }
}

确实是一个有趣的问题。答案的关键是这个Publish重载:

    // Summary:
    //     Returns an observable sequence that is the result of invoking the selector on
    //     a connectable observable sequence that shares a single subscription to the underlying
    //     sequence. This operator is a specialization of Multicast using a regular System.Reactive.Subjects.Subject`1.
    //
    // Parameters:
    //   source:
    //     Source sequence whose elements will be multicasted through a single shared subscription.
    //
    //   selector:
    //     Selector function which can use the multicasted source sequence as many times
    //     as needed, without causing multiple subscriptions to the source sequence. Subscribers
    //     to the given source will receive all notifications of the source from the time
    //     of the subscription on.
    //
    // Type parameters:
    //   TSource:
    //     The type of the elements in the source sequence.
    //
    //   TResult:
    //     The type of the elements in the result sequence.
    //
    // Returns:
    //     An observable sequence that contains the elements of a sequence produced by multicasting
    //     the source sequence within a selector function.
    //
    // Exceptions:
    //   T:System.ArgumentNullException:
    //     source or selector is null.
    public static IObservable<TResult> Publish<TSource, TResult>(this IObservable<TSource> source, Func<IObservable<TSource>, IObservable<TResult>> selector);

无论如何，希望这对你有帮助。

这不是回答，而是对你问题的澄清。

我正在努力理解用例。正如@ibebbs指出的，Distinct不是那样工作的。看起来你想要一个像DistinctLatest的东西。

这是测试用的大理石图。图中的'|'表示订阅，而不是完成。同样，假设new是一个热可观察对象，s1是一个大约在t=20的订阅者，s2是一个大约在t=1的订阅者:

   t: ------------0--------------10--------------------20------
seed: (10)(11)(12)---------------------------------------------
 new: ---------------------------(20)(21)(22)(23)(24)--(30)(31)
  s1:                                                  |(30)(31)(22)(23)(24)
  s2:              |(10)(11)(12)-(20)(21)(22)(23)(24)--(30)(31)

这是你想要的吗?

---

编辑:

来自@LeeCampbell评论的回答:

public static class RxExtensions
{
    public static IObservable<T> Cache<T, TKey>(this IObservable<T> newElements, IEnumerable<T> seedElements, Func<T, TKey> replacementSelector)
    {
        return seedElements.ToObservable()
            .Concat(newElements)
            .GroupBy(i => replacementSelector)
            .SelectMany(grp => grp.Replay(1).Publish().RefCoun‌​t());
    }
}