在 Rx 中对数组元素进行分组

Question

我想将以下实现清理成更具声明性的方法，希望有一个我缺少的 Rx 运算符。

我在流上发布了一组数据，如下所示：

var data = new[] { 'A', 'A', 'C', 'A', 'A', 'A', 'A', 'B', 'C', 'C' };

并且我想将数据中的项目分组到一个数组中直到更改，所以你的结果将是：

var result = new[] { { 'A', 'A' },  {'C' }, { 'A', 'A', 'A', 'A' }, { 'B' }, { 'C', 'C' } };

我有以下内容，我想清理 'Select' 实现。理想情况下使用 Rx 运算符（例如 GroupByUntil）。

如何改进？

 Observable.Return(data)
           .Select(items =>
           {
               var groupedItems = new List<List<char>>();

               var currentItems = new List<char>();
               var previousItem = items.FirstOrDefault();
               foreach (var currentItem in items)
               {
                   if (previousItem != currentItem)
                   {
                       groupedItems.Add(currentItems);
                        currentItems = new List<char>();
                   }

                   currentItems.Add(currentItem);
                   previousItem = currentItem;
               }
                groupedItems.Add(currentItems);
                return (IEnumerable<IEnumerable<char>>)groupedItems;
            })
            .Subscribe(groupedItems =>
            {
               foreach (var groupedItem in groupedItems)
               {
                   Console.WriteLine("{0} - {1}", groupedItem.First(), groupedItem.Count());
               }
            });

Answer 1

当输入流转换为字符流时.ToObservable。以下实现是可能的：

    var data = new[] { 'A', 'A', 'C', 'A', 'A', 'A', 'A', 'B', 'C', 'C' };

    var publishedData = data
        .ToObservable()
        .Publish()
        .RefCount();

    publishedData
        .GroupByUntil(e => e, e => e, g => publishedData.Where(i => i != g.Key))
        .Select(g => g.ToArray())
        .Merge()
        .Subscribe(groupedItems =>
        {
            Console.WriteLine("{0} - {1}", groupedItems.First(), groupedItems.Count());
        });

.GroupByUntil 运算符中的结束序列需要 .Publish().RefCount()。

在 Rx 中对数组元素进行分组

Grouping array elements in Rx

.net

c#

reactive-programming

system.reactive