如何将 IObservable 投影到 Rx 中按键分组的元素列表中

Question

给定一个IObservable(Of T)我们如何将其转换为一个IObservable(Of List(Of T))，该发出由某个键分组的元素列表？使用 GroupBy 、Select 和 Scan 运算符，我设法将源代码划分为可观察的，为每个键生成所有元素的列表。我不知道如何进一步将这些列表连接成一个列表。

    Dim source = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}.ToObservable()
    Dim keySelector = Function(element As Integer) As Integer
                          Return element Mod 3
                      End Function
    Dim result = source.GroupBy(Of Integer)(keySelector) _
                 .Select(Function(gr)
                             Return gr.Scan(New List(Of Integer), _
                                            Function(integers, current)
                                                integers.Add(current)
                                                Return integers
                                            End Function)
                         End Function)
    result.Subscribe(Sub(gr) gr.Subscribe(Sub(lst)
                                              Console.WriteLine(String.Join(",", lst))
                                          End Sub))

它产生以下输出：

1
2
3
1,4
2,5
3,6
1,4,7
2,5,8
3,6,9
1,4,7,10

虽然我需要它是：

1
1,2
1,2,3
1,4,2,3
1,4,2,5,3
1,4,2,5,3,6
1,4,7,2,5,3,6
1,4,7,2,5,8,3,6
1,4,7,2,5,8,3,6,9
1,4,7,10,2,5,8,3,6,9

Answer 1

这能满足你的需要：

Dim result = _
    Observable _
        .Create(Of List(Of Integer))( _
            Function (o)
                Dim keysFound = 0
                Dim keyOrder = New Dictionary(Of Integer, Integer)
                Return _
                    source _
                        .Do( _
                            Sub (x)
                                Dim k = keySelector(x)
                                If Not keyOrder.ContainsKey(k) Then
                                    keyOrder.Add(k, keysFound)
                                    keysFound = keysFound  + 1
                                End If
                            End Sub) _                      
                        .Scan( _
                            New List(Of Integer), _
                            Function(integers, current)
                                integers.Add(current)
                                Return integers
                            End Function) _
                        .Select(Function(integers) _
                            integers.OrderBy(Function (x) _
                                keyOrder(keySelector(x))).ToList()) _
                        .Subscribe(o)
            End Function)
result.Subscribe(Sub(gr) Console.WriteLine(String.Join(",", gr)))

我得到这个结果：

11,21,2,31,4,2,31,4,2,5,31,4,2,5,3,61,4,7,2,5,3,61,4,7,2,5,8,3,61,4,7,2,5,8,3,6,91,4,7,10,2,5,8,3,6,9

Answer 2

这不是我问题的确切答案，因为它不会产生请求的输出类型，但考虑到性能，我决定使用@Carsten建议来测试一种新方法，以使用Dictionary。我的源生成值的速度非常快，我可能不需要在值出现时进行处理，所以我可能会做一个.Throttle或一个.Sample。鉴于此，我将每个键的所有值收集到列表中，并且只发出一个Dictionary(Of Integer,Of IList(Of T))。在订阅器部分，在限制或采样后，收到的字典被平展。

    Dim source = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}.ToObservable()
    Dim keySelector = Function(element As Integer) As Integer
                          Return element Mod 3
                      End Function
    Dim result = source.Select(Function(i)
                                   Return New With {.reminder = keySelector(i), .value = i}
                               End Function) _
                        .Scan(ImmutableDictionary(Of Integer, IImmutableList(Of Integer)).Empty, _
                              Function(accumulate, current)
                                  Dim builder = accumulate.ToBuilder()
                                  If Not builder.ContainsKey(current.reminder) Then
                                      builder.Add(current.reminder, ImmutableList(Of Integer).Empty)
                                  End If
                                  Dim currentList = builder(current.reminder)
                                  builder(current.reminder) = currentList.Add(current.value)
                                  Return builder.ToImmutable()
                              End Function)

    result.Throttle(TimeSpan.FromMilliseconds(100)) _
        .Subscribe(Sub(dictionary)
                       Console.WriteLine( _
                           String.Join(",", dictionary.SelectMany(Function(pair)
                                                                      Return pair.Value
                                                                  End Function)))
                   End Sub)

它产生这个：

3,6,9,1,4,7,10,2,5,8

有趣的是，ImmutableDictionary(Of Tkey, TValue)在添加键时订购键，而Dictionary(Of TKey, TValue)则没有。目前，这不是一个真正的问题。