Группировка элементов массива в 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' } }; 

У меня есть следующее, и я хочу очистить реализацию «Выбрать». В идеале используется оператор 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()); 
     }); 

.Publish().RefCount() необходима для последовательности закрытия в операторе .GroupByUntil.