CODEWITHSUNDEEP
c#.netsystem.reactivereactive-programming
AwkwardCoder
AwkwardCoder

Reputation: 25631

Grouping array elements in Rx

I'm want to clean-up the following implementation into a more declarative approach, hopefully there is an Rx operator I'm missing.

I have an array of data published on a stream, something like this:

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

And I want to group items in the data into a array until the change, so you the result would be:

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

I have the following, and I want to clean-up the 'Select' implementation. Ideally using an Rx operator (e.g. GroupByUntil).

How can this be improved?

 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());
               }
            });

Upvotes: 1

Views: 132

Answers (1)

Lakerfield
Lakerfield

Reputation: 1091

When the input stream is converted to a stream of characters with .ToObservable. The following implementation is possible:

    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());
        });

The .Publish().RefCount() is needed for the closing sequence in the .GroupByUntil operator.

Upvotes: 3

Related Questions