Observed <IEnumerable> Gets Differences Between Sequence Elements

Question

I have an observable that emits a sequence of IEnumerables

1: [1,2,3,4]

2: [1,3,4]

3: [1,5,6]

etc..

I want to try to create two observables from this:

• One that emits IEnumerable from newly added elements:

1: [1,2,3,4]

2: []

3: [5, 6]

etc..

• One that emits IEnumerable from recently deleted items:

1: []

2: [2]

3: [3,4]

etc..

Is there a way to do this using System.Reactive without relying on maintaining a separate data structure to compare changes with?

Answer 1

It is quite simple if you use Observable.Zipand Enumerable.Exceptto easily compare the element n with the element n-1.

public static class IObservableIEnumerableExtensions
{
    public static IObservable<IEnumerable<T>> GetAddedElements<T>(this IObservable<IEnumerable<T>> source)
    {
        return source.Zip(source.StartWith(Enumerable.Empty<T>()), (newer, older) => newer.Except(older));
    }

    public static IObservable<IEnumerable<T>> GetRemovedElements<T>(this IObservable<IEnumerable<T>> source)
    {
        return source.Zip(source.StartWith(Enumerable.Empty<T>()), (newer, older) => older.Except(newer));
    }
}

And here is the runner’s code:

var source = new Subject<IEnumerable<int>>();

var addedElements = source.GetAddedElements();
var removedElements = source.GetRemovedElements();

addedElements.Dump();   //Using Linqpad
removedElements.Dump(); //Using Linqpad

source.OnNext(new int[] { 1, 2, 3, 4 });
source.OnNext(new int[] { 1, 3, 4 });
source.OnNext(new int[] { 1, 5, 6 });

Answer 2

, - , .

public static IObservable<IEnumerable<T>> CumulativeAdded<T>(this IObservable<IEnumerable<T>> src) {
    var memadd = new HashSet<T>();

    return src.Select(x => x.Where(n => memadd.Add(n)));
}

public static IObservable<IEnumerable<T>> CumulativeRemoved<T>(this IObservable<IEnumerable<T>> src) {
    var memdiff = new HashSet<T>();

    return src.Select(x => { foreach (var n in x) memdiff.Add(n); return memdiff.AsEnumerable().Except(x); });
}

}