Observing items being added to a list in RX

Question

I am trying to get a simple demo working.

I have a collection of strings and I want to watch it for additions, without using any control event code. Somehow I've gotten the impression, perhaps wrongly, that Rx or another part of .Net supports this, without resorting to wiring up all the various events that (may or may not) add members to the collection.

If I replace my source with an interval, as in the commented out code, the delegate gets called (ala, var source = Observable.Interval(TimeSpan.FromSeconds(1));. This gives me hope I can do what I want to here, perhaps falsely.

The basic example came from Creating and Subscribing to Simple Observable Sequences

In the code below, what I want to do is observe the source collection directly (not through control events) and call a delegate when an item is added to the collection.

I would prefer not to bypass LINQ or RX if they do, indeed, support my thesis that they support this functionality.

using Microsoft.VisualBasic;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Data;
using System.Diagnostics;
using System.Collections.ObjectModel;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;

public class frmRx
{

    ObservableCollection<string> ObservableCollection = new ObservableCollection<string>();
    Dim source = Observable.ToObservable(ObservableCollection).ObserveOn(SynchronizationContext.Current)

    //this code worked, but it's not a collection I made
    //source = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => x.ToString).ObserveOn(SynchronizationContext.Current);

    private void frmRx_Load(System.Object sender, System.EventArgs e)
    {
        IConnectableObservable<string> Publication = Observable.Publish<string>(source);
        Publication.Subscribe(x => { AddToTreeView(x); }, ex => { }, () => { });
        Publication.Connect();
    }

    private void AddToTreeView(string Text)
    {
        TreeView1.Nodes.Add(Text); //this never gets called
    }

    // this is just my test way of adding a member to the collection.
    // but the adding could happen anywhere,
    // and I want to watch the collection changes regardless of how it came about
    private void TextBox1_TextChanged(System.Object sender, System.EventArgs e)
    {
        ObservableCollection.Add(TextBox1.Text.Last);
    }
    public frmRx()
    {
        Load += frmRx_Load;
       TextBox1.TextChanged += TextBox1_TextChanged;
        }
    }

Answer 1

You are making the mistake that Observable.ToObservable(ObservableCollection) will create an IObservable<string> that will produce values for future updates of the ObservableCollection.

It does not.

The .ToObservable(...) extension method simply turns an IEnumerable<> into and IObservable<> such that the values are enumerated at the moment in time that the observable is subscribed to.

You need to use a Subject<string> if you want new values pushed to your subscribers.

Further to this, your code isn't as simple as possible. Why are you mucking around with the publishing of observables?

Here's the simplest code you could write to get this to work:

public class frmRx
{
    private Subject<string> source = new Subject<string>();

    public frmRx()
    {
        source.ObserveOn(this).Subscribe(x => TreeView1.Nodes.Add(x));
        TextBox1.TextChanged += (s, e) => source.OnNext(TextBox1.Text);
    }
}

I've included the .ObserveOn(this) as it is a good habit to put them in but it shouldn't be needed in this case.

What I would prefer to see is this though:

public class frmRx
{
    public frmRx()
    {
        Observable
            .FromEventPattern(
                h => textBox1.TextChanged += h,
                h => textBox1.TextChanged -= h)
            .Select(x => textBox1.Text)
            .Subscribe(x => treeView1.Nodes.Add(x));
    }
}

This avoids subjects and it is as simple as it can get while being strongly-typed.

It would even be better to go one step further and do a better job of cleaning up your subscription on closing, like this:

        var subscription =
            Observable
                .FromEventPattern(
                    h => textBox1.TextChanged += h,
                    h => textBox1.TextChanged -= h)
                .Select(x => textBox1.Text)
                .Subscribe(x => treeView1.Nodes.Add(x));

        this.FormClosing += (s, e) => subscription.Dispose();

Answer 2

The easiest way is to use an Rx.net framework, like ReactiveUI. I shall show you how that would work.

ReactiveList<string> _collection = new ReactiveList<string>();

public Constructor()
{
     var subscription = _collection.ItemsAdded.Subject(added => doSomething(added));
     this.Disposed += (o, e) => subscription.Dispose();
}

Here how much code you need to get this to "work" without third party. Also, this code has NO exception handling in it at all.

public class ViewModel
{
    private ObservableCollection<string> _source = new ObservableCollection<string>();

    public ViewModel()
    {
        //There will be issues with the exception handling
        Observable.FromEventPattern
            <NotifyCollectionChangedEventHandler, NotifyCollectionChangedEventArgs>
            (x => _source.CollectionChanged += x, x => _source.CollectionChanged -= x)
            .Where(x => x.EventArgs.Action == NotifyCollectionChangedAction.Add)
            .SelectMany(x => x.EventArgs.NewItems.Cast<string>())
            .Subscribe(AddToTreeView);

    }

    public void AddToTreeView(string text)
    {
        TreeView1.Nodes.Add(text);
    }
}