Merging two observable collection and binding to the Listbox using Rx

I need to merge 2 ObservableCollection into one and bind it to the Grid and need realtime updates to flow to the Grid. for .e.g.

ObservableCollection<int> First = new ObservableCollection<int>开发者_运维技巧;();
ObservableCollection<int> Second  = new ObservableCollection<int>();

//Some Rx Psuedo Code (p.s. this is not the actual code, this is where i need help)
{
    var guicollection = First
        .Where(i => i%2)
        .Merge(Second.Where(i => i % 3)).ToCollection();
}

listBox1.ItemsSource = guidcollection;

First.Add(1);
First.Add(2);
First.Add(3);
First.Add(4);
First.Add(5);
Second.Add(1);
Second.Add(2);
Second.Add(3);
Second.Add(4);

// Now the guicollection should have the following items 2,4 from FirstCollection
// and 3 from second collection

So the above guicollection should work realtime wheneve an object is added to the first or second collection the filtering should be applied and the filtered items should be added to the guicollection. I read somewhere that Rx framework can really help here. Please help me replace the Psudeo code above with the actual Rx Code. Thanks.

---

Here's my solution for you:

Func<ObservableCollection<int>,
    Func<int, bool>,
    IObservable<int>> getAddsWhere =
        (oc, pred) =>
            from ep in Observable
                .FromEventPattern<NotifyCollectionChangedEventHandler,
                    NotifyCollectionChangedEventArgs>(
                        h => oc.CollectionChanged += h,
                        h => oc.CollectionChanged -= h)
            where ep.EventArgs.Action == NotifyCollectionChangedAction.Add
            from i in ep.EventArgs.NewItems.OfType<int>()
            where pred(i)
            select i;

var firsts = getAddsWhere(First, i => i % 2 == 0);
var seconds = getAddsWhere(Second, i => i % 3 == 0);

var boths = firsts.Merge(seconds);

boths.Subscribe(i => guicollection.Add(i));

I tested it and it works as you requested - 2, 3 & 4 end up in guicollection.

---

EDIT: Changed to show how to handle all of the NotifyCollectionChangedAction enum values.

The NotifyCollectionChangedAction enum has five values:

1. Add
2. Move
3. Remove
4. Replace
5. Reset

There's nothing to do for Move - it's just an internal operation.

The NewItems collection on NotifyCollectionChangedEventArgs contains values for Add & Replace.

The OldItems collection on NotifyCollectionChangedEventArgs contains values for Remove & Replace.

The tricky operation is Reset - which occurs when Clear() is called on the collection - because it doesn't tell you which items were clears and then items have already been cleared by the time the event is raised.

So the only solution is to create an extension method that returns IObservable<ObservableCollectionOperation<T>> and internally tracks the changes so that a series of removes can be issued when Clear is called.

Before I dump a lot of code here, I'll show you what the calling code looks like. It's quite simple and straight forward.

var FirstOps = First.ToOperations(i => i % 2 == 0);
var SecondOps = Second.ToOperations(i => i % 3 == 0);

var BothOps = FirstOps.Merge(SecondOps);

var subscription = BothOps.Subscribe(guicollection);

Very neat, huh?

The class ObservableCollectionOperation<T> is defined like so:

public class ObservableCollectionOperation<T>
{
    public readonly T Value;
    public readonly Operation Operation;

    public static ObservableCollectionOperation<T> Add(T value)
    {
        return new ObservableCollectionOperation<T>(value, Operation.Add);
    }

    public static ObservableCollectionOperation<T> Remove(T value)
    {
        return new ObservableCollectionOperation<T>(value, Operation.Remove);
    }

    public ObservableCollectionOperation(T value, Operation operation)
    {
        this.Value = value;
        this.Operation = operation;
    }

    public override int GetHashCode()
    {
        return this.Value.GetHashCode()
            * (this.Operation == Operation.Add ? 1 : -1);
    }

    public override bool Equals(object obj)
    {
        if (obj is ObservableCollectionOperation<T>)
        {
            var other = obj as ObservableCollectionOperation<T>;
            return this.Value.Equals(other.Value)
                    && this.Operation.Equals(other.Operation);
        }
        return false;
    }
}

The Operation enum is required to differentiate between adding and removing items and it unsurprisingly looks like this:

public enum Operation
{
    Add,
    Remove,
}

Now for the extension method.

public static IObservable<ObservableCollectionOperation<T>>
    ToOperations<T>(this ObservableCollection<T> @this)
{
    return Observable.Create<ObservableCollectionOperation<T>>(o =>
    {
        var local = new List<T>(@this);

        Func<NotifyCollectionChangedEventArgs,
            ObservableCollectionOperation<T>[]>
                getAdds = ea =>
                {
                    var xs = new T[] { };
                    if (
                        ea.Action == NotifyCollectionChangedAction.Add
                        || ea.Action == NotifyCollectionChangedAction.Replace)
                    {
                        xs = ea.NewItems.Cast<T>().ToArray();
                        local.AddRange(xs);
                    }
                    return xs
                        .Select(x =>
                            ObservableCollectionOperation<T>.Add(x))
                        .ToArray();
                };

        Func<NotifyCollectionChangedEventArgs,
            ObservableCollectionOperation<T>[]>
                getRemoves = ea =>
                {
                    var xs = new T[] { };
                    if (
                        ea.Action == NotifyCollectionChangedAction.Remove
                        || ea.Action == NotifyCollectionChangedAction.Replace)
                    {
                        xs = ea.OldItems.Cast<T>().ToArray();
                        Array.ForEach(xs, x => local.Remove(x));
                    }
                    return xs
                        .Select(x =>
                            ObservableCollectionOperation<T>.Remove(x))
                        .ToArray();
                };

        Func<NotifyCollectionChangedEventArgs,
            ObservableCollectionOperation<T>[]>
                getClears = ea =>
                {
                    var xs = new T[] { };
                    if (ea.Action == NotifyCollectionChangedAction.Reset)
                    {
                        xs = local.ToArray();
                        local.Clear();
                    }
                    return xs
                        .Select(x =>
                            ObservableCollectionOperation<T>.Remove(x))
                        .ToArray();
                };

        var changes =
            from ep in Observable
                .FromEventPattern<NotifyCollectionChangedEventHandler,
                    NotifyCollectionChangedEventArgs>(
                        h => @this.CollectionChanged += h,
                        h => @this.CollectionChanged -= h)
            let adds = getAdds(ep.EventArgs)
            let removes = getRemoves(ep.EventArgs)
            let clears = getClears(ep.EventArgs)
            from x in clears.Concat(removes).Concat(adds).ToObservable()
            select x;

        return changes.Subscribe(o);
    });
}

I added an overloaded extension method to help with filtering:

public static IObservable<ObservableCollectionOperation<T>>
    ToOperations<T>(
        this ObservableCollection<T> @this,
        Func<T, bool> filter)
{
    return @this.ToOperations().Where(op => filter(op.Value));
}

And finally I created a helper method to allow the observable operations to be played into an "observer" ObservableCollection<T>:

public static IDisposable
    Subscribe<T>(
        this IObservable<ObservableCollectionOperation<T>> @this,
        ObservableCollection<T> observer)
{
    return @this.Subscribe(op =>
    {
        switch (op.Operation)
        {
            case Operation.Add :
                observer.Add(op.Value);
                break;
            case Operation.Remove :
                observer.Remove(op.Value);
                break;
        }
    });
}

Now, yes, this does handle removes and it works on the sample operation that you gave. :-)

---

I don't know anything about the Rx Framework, however ObservableCollections notify the UI anytime the collection's contents change, so you should only need to add/remove items from your bound collection to have the UI update

The merge can be done with a script such as the following:

public ObservableCollection<object> MergeCollections(
    ObservableCollection<object> first,
    ObservableCollection<object> second)
{
    foreach(var item in second)
    {
        if (!(first.Contains(item)))
            first.Add(item);
    }

    return first;
}