Using Rx (Reactive Extensions) to watch for specific item in ObservableCollection

Question

I have an ObservableCollection that I need to reference for a specific item. If the item is not there, I need to monitor it via Reactive Extensions for if/when the items appears, but need some help in setting up the statement. I'm still unfamiliar with how all the different Linq extensions are intended to work, so I'm not sure how to do this. Can anyone point me in the right direction?

To illustrate better, I need to something like the following:

public class myitem :INotifyPropertyChanged
{
    private string _key;
    private string _value;

    public string key
    {
        get { return _key; }
        set { _key = value; NotifyPropertyChanged("key"); }
    }

    public string myvalue
    {
        //proper getter/setter, blah, blah
    }
}

ObservableCollection<myitem> _collection = mycollection;
var x = Observable.FromEvent<NotifyCollectionChangedEventHandler, NotifyCollectionChangedEventArgs>(
    h => new NotifyCollectionChangedEventHandler(h),
    h => _collection.CollectionChanged += h,
    h => _collection.CollectionChanged -= h);

string keywaitingfor = "thiskey";
string valuewaitingfor = x.Where(xx => xx.key == keywaitingfor).First().myvalue;

This isn't exactly my scenario, but hopefully you can see what I'm trying to do. The ObservableCollection may contain no items to begin, and the property values come in asyncronously. I know the last line isn't right, I need to have an Observable on the class PropertyChanged event within a lambda... but am still confused about how to just get that valuewaiting for when both conditions are met.

Answer 1

The generic ObservableCollection has nothing to do with the IObservable interface. You can however monitor an ObservableCollection's CollectionChanged event through Rx using the following:

ObservableCollection<SomeType> items = yourObserableCollection;
var itemAddedObservable = Observable
         .FromEventPattern<NotifyCollectionChangedEventArgs>(items, "CollectionChanged")
         .Select(change => change.EventArgs.NewItems)

This will give you a notification whenever item(s) are added to the ObservableCollection. The items will be a non-generic IList, so we can cast that to an IEnumerable of the underlying SomeType and SelectMany on that .AsObservable to get a new observable stream of the incoming values. Finally in the Subscribe, you do what you want with the final value (rather than using the blocking First call):

var filteredAddedItem = from added in itemAddedObservable
                        from itemAdded in added.OfType<SomeType>().ToObservable()
                        where itemAdded.key = keywaitingfor
                        select itemAdded;

var sub = filteredAddedItem.Subscribe(item => Console.WriteLine("Received " + item.myvalue));

Answer 2

Add ReactiveUI to your project, then you can use the ReactiveCollection class, which derives from WPF's ObservableCollection. Then this is easy as pie:

theCollection.ItemsAdded
    .Where(x => x.Key == "Foo")
    .Subscribe(x => Console.WriteLine("Received Item!"));