The Use of Observable in General

Question

Although I use Observable and Rx in my code from time to time I still have this question in terms of "push model" and its use.

For example, let's say I have this simple code:

  private readonly static List<string> numbers = new List<string>
  {
      "1",
      "2",
      "3",
      "4",
      "5"
  };

  static void Main(string[] args)
  {
      PrintCollection();
      Console.ReadLine();
      numbers.Add("6");
      Console.ReadLine();
  }

  private static void PrintCollection()
  {
      IObservable<string> observable = numbers.ToObservable();
      observable.Subscribe<string>(x => { Console.WriteLine(x); });
  }

When the program runs only 1-5 will be printed but not 6 unless I use something like an ObservableCollection and hook up the "CollectionChanged" event. However, this makes me wonder what the "push model" really is. I always thought the "push model" meant once the data (colleciton) has subscribed to an event all newly added data to that collection would be subscribed to the same event as well.

Also, most of the examples I've seen using Observable seem to be WPF, view-model-driven implmenentation and I wonder if anyone has used it for any back-end processing and what would be a typical example of that?

JerKimball JerKimball · Accepted Answer · 2013-04-01T22:55:17

I think you are confusing what ToObservable does - basically, it says "see this set of things? I want to create a new stream that feeds these values back to me." It doesn't "wrap" the source list in any sort of meaningful way such as to listen to future changes, it's more of a snapshot in time - the resulting observable is what that list looked like when the observable was created.

There are numerous ways to model a "live view" of a collection - one you've mentioned: create an ObservableCollection and create the observable from writing into the CollectionChanged event. You can also use a Subject here, although it injects the concept of state into the "stateless" world of rx.

Actually, that's a good point to emphasize: rx streams are far more "functional" in concept than traditional imperative/oo programming; you declare a stream of potential values over time, not a state machine (although you can force it to act like a state machine)...you figure out what the values you want/need "look"over time and that's how you declare your stream.

In a specific case like this, however, declaring a stream of values that vary randomly is difficult; that is better satisfied by other means (eventing, polling, etc).

That said, you could write something like this:

var numbers = new List<string>
{
    "1",
    "2",
    "3",
    "4",
    "5"
};
var source = new ObservableCollection<string>(numbers);
var query = Observable.Create<string>((obs) =>
    {
        foreach(var oldItem in source)
        {
            obs.OnNext(oldItem);
        }
        NotifyCollectionChangedEventHandler h;
        h = (o, e) => 
        {
            Console.WriteLine("Collection changed!");
            foreach(var item in e.NewItems)
            {
                obs.OnNext(item as string);
            }
        };
        source.CollectionChanged += h;
        return Disposable.Create (() => source.CollectionChanged -= h);
    });

using(query.Subscribe(Console.WriteLine))
{
    source.Add("6");
    Console.ReadLine();
}

Output:

1
2
3
4
5
Collection changed!
6

The Use of Observable in General

2 Answers