Rx_Observer_Observables.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;

namespace ConsoleApplication3
{
    class Program2
    {
        static void Main(string[] args)
        {
            Console.WriteLine("-------------ONE SUBSCRIBER-------------");

            var producer = new Producer();
            var c = new ConsoleConsumer();

            var unsubscribe = producer.Subscribe(c);
            producer.Start();
            unsubscribe.Dispose();

            Console.WriteLine("-------------MULTIPLE SUBSCRIBERS-------------");

            var u = new ConsoleConsumer();

            var connected = producer.Publish();
            connected.Subscribe(c);
            connected.Subscribe(u);
            connected.Connect();
            producer.Start();
            producer.Dispose();

            

            Console.WriteLine("-------------COMPOSITION SUBSCRIBERS-------------");

            producer = new Producer();
            
            var k = new ProducerSequenceChain(producer, new ConsoleConsumer(), new DebugConsumer());
            producer.Start();

            Console.ReadLine();
        }

        class ProducerSequenceChain
        {
            public IObservable<string> Trigger { get; internal set; }
            public IObserver<string>[] Results { get; internal set; }

            public ProducerSequenceChain(IObservable<string> trigger, params IObserver<string>[] results)
            {
                Trigger = trigger;
                Results = results;                
                results.Run(result => Trigger.Subscribe(result));
            }
        }


        class Producer : IObservable<string>, IDisposable
        {
            private readonly List<IObserver<string>> _observers;

            public Producer()
            {
                _observers = new List<IObserver<string>>();
            }

            public void Start()
            {
                foreach (var o in this._observers)
                {
                    o.OnNext("peter");
                    o.OnNext("paul");
                    o.OnNext("maria");
                }
            }

            public IDisposable Subscribe(IObserver<string> observer)
            {
                if (!_observers.Contains(observer))
                {
                    _observers.Add(observer);
                }
                return this;
            }

            public void Dispose()
            {
                _observers.RemoveAll(x => true);
            }
        }

        class ConsoleConsumer : IObserver<string>
        {

            public void OnNext(string value)
            {
                Console.WriteLine(value);
            }

            public void OnError(Exception error)
            {
                Console.WriteLine("finished.");
            }

            public void OnCompleted()
            {
                Console.WriteLine("finished.");
            }
        }

        class DebugConsumer : IObserver<string>
        {

            public void OnNext(string value)
            {
                Debug.WriteLine(value);
            }

            public void OnError(Exception error)
            {
                Debug.WriteLine("finished.");
            }

            public void OnCompleted()
            {
                Debug.WriteLine("finished.");
            }
        }

    }
}