Grouping an IObservable and filtering out those with a count greater than one

Question

I have this code:

IObservable<Provider> allProviders = _context.ReadAllProviders();

Provider[] results = allProviders
    .ToEnumerable()
    .GroupBy(
        p => p.Name,
        (name, group) =>
        {
            var array = group.ToArray();
            return array.Length == 1 ? array[0] : null;
        },
        StringComparer.OrdinalIgnoreCase)
    .Where(p => p != null)
    .ToArray();

I'm filtering out providers that aren't unique on the Name property, but I'd like to do perform the GroupBy on the IObservable instead of an IEnumerable as I intend to do more observable stuff on it afterwards. I'm having trouble with the syntax when it comes to filtering the grouped observable:

IObservable<Provider> allProviders = _context.ReadAllProviders();

IObservable<Provider> results = allProviders
    .GroupBy(p => p.Name, StringComparer.OrdinalIgnoreCase)
    .Where(group =>
    {
        IObservable<int> count = group.Count();

        return count == 1; // this doesn't compile obviously. I'm not sure where to go from here.
    })
    .Merge();

Answer 1

Logically, you cannot know how many members will be in an Observable group until the source stream completes - since any element could potentially contribute to any of the groups.

Ergo, you cannot filter the groups by their count until the source stream is finished, and you must consent to cache the results and then process them.

In other words, IEnumerable<T> is the way to go here. Sure, you could turn it back into an IObservable<T>, but there's probably no point.

As an aside, if you do ever need to filter groups, do it in a Select clause as in this simple example. It won't work for Count() though, for the reasons mentioned above:

var source = Observable.Range(0, 10);
source.GroupBy(n => n < 5)        
      .Select(x => x.Where(y => y % 2 == 0));

Answer 2

I think this works to produce a stream of unique named Provider objects.

IObservable<Provider> results = allProviders
    .GroupBy(p => p.Name)
    .SelectMany(xs => xs
        .Skip(1)
        .Take(1)
        .Zip(xs, (x1, x0) => x0))
    .Where(p => p != null);