RxでDistinctLatest（およびキャッシング）演算子を実装するにはどうすればよいですか？

Question

質問がありましたA cache serving updates and new values as “DistinctLatest” and full cache contents upon subscription、これはコミュニティによってうまく処理されました。前述の質問で定義されたような値のキャッシングと置き換えの実際の目標は、.DistinctLatest演算子で定義できるという質問が提起されました。

OK！そのような演算子についてはあまり話がないようです。検索しながら考えているうちに、私はReactiveX: Group and Buffer only last item in each groupを見つけました。これは一種のものです。オリジナルの問題を模倣するために、私は

/// <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 seed elements when the cache is started.</param> 
/// <param name="replacementSelector">The replacement selector to choose 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 s = newElements.StartWith(seedElements).GroupBy(replacementSelector).Select(groupObservable => 
    { 
     var replaySubject = new ReplaySubject<T>(1); 
     groupObservable.Subscribe(value => replaySubject.OnNext(value)); 

     return replaySubject; 
    }); 

    return s.SelectMany(i => i);    
} 

しかし、どちらかのトリックを行うようには見えないテストを行うと、キャッシング演算子を書き込もうとしました。初めに購読していて、初期値と更新（そして新しい値）が観察されたようです。最後に購読している場合は、置き換えられたシード値のみが記録されます。

ここでは、一般的なDistinctLast演算子について考えていますが、これはうまく動作しません。この "キャッシュ"が追加するのは、シード値とグループの平坦化ですが、テストではそうではありません。私もグループ化していくつかのことを試しましたが、.TakeLast()もありますが、ダイスはありません。

誰かがポインタを持っていたり、これについて熟考していればうれしいことですが、これは一般的に有益なものであることが分かりました。

Answer 1

@ LeeCampbellはこれのためにほとんどの作業を行いました。他の参照される質問を参照してください。とにかく、コードは

public static class RxExtensions 
{ 
    public static IObservable<T> DistinctLatest<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()); 
    } 
}