Rx如何按鍵複合一個複合對象，然後在沒有「停止」流的情況下執行SelectMany？

Question

這與我的另一個問題here有關。 James World提出的解決方案如下：

// idStream is an IObservable<int> of the input stream of IDs 
// alarmInterval is a Func<int, TimeSpan> that gets the interval given the ID 
var idAlarmStream = idStream 
.GroupByUntil(key => key, grp => grp.Throttle(alarmInterval(grp.Key))) 
.SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key))); 

<編輯2：

問：我如何立即啓動定時器，而無需等待第一個事件的時候？這是我的問題的根本問題，我猜。爲此，我計劃發送帶有我知道應該在那裏的ID的虛擬對象。但正如我在下面寫的，我結束了一些其他問題。不過，我認爲解決這個問題也很有趣。

然後轉發其他有趣的部分！現在，如果我想組像下面和組通過如下關鍵一個複雜的對象（不會編譯）

var idAlarmStream = idStream 
    .Select(i => new { Id = i, IsTest = true }) 
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key))) 
    .SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key))); 

然後我惹上麻煩。我無法修改關於SelectMany，Concat和Observable.Return的部分，以便查詢可以像以前一樣工作。舉例來說，如果我做查詢作爲

var idAlarmStream = idStream 
    .Select(i => new { Id = i, IsTest = true }) 
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key))) 
    .SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key.First()))) 
    .Subscribe(i => Console.WriteLine(i.Id + "-" + i.IsTest); 

然後需要兩個事件都可以在Subscribe觀察的輸出之前。我收集的是First的調用效果。此外，我還希望在調用alarmInterval時也使用複雜的對象屬性。

有人可以提供解釋發生了什麼，甚至可能是解決方案嗎？使用未經修改的解決方案的問題在於，分組不會僅將Ids看作關鍵值，而且還會看到IsTest字段。

<編輯：作爲一個說明，這個問題大概可以firsly通過創建一個明確的類或結構解決，那麼IEquatable其次實現了自定義，然後用詹姆斯的代碼，那麼該分組將被髮生身份證。雖然感覺像黑客。

Answer 1

另外，如果你要計算你看到一個項目的次數警報響起之前，你可以像這樣做，在Select採取反超載的優勢。

var idAlarmStream = idStream 
    .Select(i => new { Id = i, IsTest = true }) 
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key)) 
    .SelectMany(grp => grp.Select((count, alarm) => new { count, alarm }).TakeLast(1)); 

注意，對於第一個（種子）項目，這將是0--這可能是你想要的。

Answer 2

您正在您的選擇中創建一個匿名類型。讓我們稱它爲A1。我會假設你的idStream是一個IObservable。由於這是GroupByUntil中的關鍵，所以您不必擔心密鑰比較 - int等於很好。

GroupByUntil是IObservable<IGroupedObservable<int, A1>>。

書面的是的SelectMany試圖成爲一個IObservable<A1>。這裏您只需要Concat(Observable.Return(grp.Key)) - 但Key的類型和Group元素的類型必須匹配，否則SelectMany將不起作用。所以關鍵也必須是A1。匿名類型使用結構相等性，返回類型將是A1的流 - 但您不能將其聲明爲公共返回類型。

如果你只是想的ID，您應該添加一個.Select(x => x.Id)後Throttle：

var idAlarmStream = idStream 
    .Select(i => new { Id = i, IsTest = true }) 
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key) 
              .Select(x => x.Id)) 
    .SelectMany(grp => grp.IgnoreElements().Concat(Observable.Return(grp.Key))); 

如果你想A1代替 - 你需要創建一個實現平等的具體類型。

編輯

我沒有測試它，但你也可以將其壓平更簡單地說是這樣，我覺得這是更容易！它輸出A1雖然，所以你必須處理，如果你需要返回流的地方。

var idAlarmStream = idStream 
    .Select(i => new { Id = i, IsTest = true }) 
    .GroupByUntil(key => key.Id, grp => grp.Throttle(alarmInterval(grp.Key)) 
    .SelectMany(grp => grp.TakeLast(1));