Rx: Observing IEnumerable for DataMapping
I am trying Rx to enable a pipeline of actions to be executed upon enumeration of a sequence. I know that this can be done with plain old .NET events, but I wanted to arrive at an Rx solution.

So I have this method which will correctly map and notify the observer passed in.

        public IEnumerable<Person> Map<T>(IEnumerable<T> reader, IObserver<Tuple<T, Person>> observer) {
            var enumerator = reader.GetEnumerator();
            while (enumerator.MoveNext()) {

                var item = default(Person);
                try {
                    var current = enumerator.Current;
                    dynamic record = enumerator.Current;
                    item = new Person { FirstName = record.FirstName };
                    var tuple = Tuple.Create(current, item);
                    observer.OnNext(tuple);
                } catch (Exception ex) {
                    observer.OnError(ex);
                }
                yield return item;
            }
            observer.OnCompleted();
            yield break;
        }

I want to enable multiple observers and the following signature:

    public IObservable<Tuple<T,Person>> Map<T>(IEnumerable<T> reader);

With this signature I can supply as many observers as I want but I still want to build the enumerator. How would I go about this?

Note: I know I can supply multiple observers in my original signature (i.e. passing in a Subject that has multiple subscribers, but the core of the question is how to return this. I've also looked at this [series of articles][1], but I think I'm missing something.


  [1]: http://blogs.msdn.com/b/jeffva/archive/2010/07/26/rx-on-the-server-part-2-of-n-asynchronous-streamreader.aspx
Rx: Observing IEnumerable for DataMapping
===

I am trying Rx to enable a pipeline of actions to be executed upon enumeration of a sequence. I know that this can be done with plain old .NET events, but I wanted to arrive at an Rx solution.

So I have this method which will correctly map and notify the observer passed in.

        public IEnumerable<Person> Map<T>(IEnumerable<T> reader, IObserver<Tuple<T, Person>> observer) {
            var enumerator = reader.GetEnumerator();
            while (enumerator.MoveNext()) {

                var item = default(Person);
                try {
                    var current = enumerator.Current;
                    dynamic record = enumerator.Current;
                    item = new Person { FirstName = record.FirstName };
                    var tuple = Tuple.Create(current, item);
                    observer.OnNext(tuple);
                } catch (Exception ex) {
                    observer.OnError(ex);
                }
                yield return item;
            }
            observer.OnCompleted();
            yield break;
        }

I want to enable multiple observers and the following signature:

    public IObservable<Tuple<T,Person>> Map<T>(IEnumerable<T> reader);

With this signature I can supply as many observers as I want but I still want to build the enumerator. How would I go about this?

Note: I know I can supply multiple observers in my original signature (i.e. passing in a Subject that has multiple subscribers, but the core of the question is how to return this. I've also looked at this [series of articles][1], but I think I'm missing something.


  [1]: http://blogs.msdn.com/b/jeffva/archive/2010/07/26/rx-on-the-server-part-2-of-n-asynchronous-streamreader.aspx