Ping Pong Actors with Rx

RxPingPong.cs

using System;
using System.Collections.Generic;
using System.Linq;

namespace RxPingPong
{
    /// <summary>Simple Ping Pong Actor model using Rx</summary>
    /// <remarks>
    /// You'll need to install the Reactive Extensions (Rx) for this to work.
    /// You can get the installer from <see href="http://msdn.microsoft.com/en-us/devlabs/ee794896.aspx"/>
    /// </remarks>
    class Program
    {
        static void Main(string[] args)
        {
            var pong = new Pong();
            var ping = new Ping(10);

            var pongSubscription = ping.Subscribe(pong);
            var pingSubscription = pong.Subscribe(ping);

            Console.ReadKey();

            pingSubscription.Dispose();
            pongSubscription.Dispose();
        }
    }

    class Ping : ISubject<int>
    {
        private readonly int _iterations;

        /// <summary>
        /// Initializes a new instance of the <see cref="Ping"/> class.
        /// </summary>
        /// <param name="iterations">The iterations.</param>
        public Ping(int iterations)
        {
            _iterations = iterations;
        }

        #region Implementation of IObserver<int>

        /// <summary>
        /// Notifies the observer of a new value in the sequence.
        /// </summary>
        public void OnNext(int value)
        {
            Console.WriteLine("Ping received Pong.");
        }

        /// <summary>
        /// Notifies the observer that an exception has occurred.
        /// </summary>
        public void OnError(Exception exception)
        {
            Console.WriteLine("Ping experienced an exception and had to quit playing.");
        }

        /// <summary>
        /// Notifies the observer of the end of the sequence.
        /// </summary>
        public void OnCompleted()
        {
            Console.WriteLine("Ping finished.");
        }

        #endregion

        #region Implementation of IObservable<int>

        /// <summary>
        /// Subscribes an observer to the observable sequence.
        /// </summary>
        public IDisposable Subscribe(IObserver<int> observer)
        {
            return Enumerable
                .Range(1, _iterations)
                .Reverse()
                .ToObservable()
                .Asynchronous()
                .Subscribe(observer);
        }

        #endregion

        #region Implementation of IDisposable

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        public void Dispose()
        {
            OnCompleted();
        }

        #endregion
    }

    class Pong : ISubject<int>
    {
        #region Implementation of IObserver<int>

        /// <summary>
        /// Notifies the observer of a new value in the sequence.
        /// </summary>
        public void OnNext(int value)
        {
            Console.WriteLine("Pong received Ping.");
            // Need to publish a response.
        }

        /// <summary>
        /// Notifies the observer that an exception has occurred.
        /// </summary>
        public void OnError(Exception exception)
        {
            Console.WriteLine("Pong experienced an exception and had to quit playing.");
        }

        /// <summary>
        /// Notifies the observer of the end of the sequence.
        /// </summary>
        public void OnCompleted()
        {
            Console.WriteLine("Pong finished.");
        }

        #endregion

        #region Implementation of IObservable<Pong>

        /// <summary>
        /// Subscribes an observer to the observable sequence.
        /// </summary>
        public IDisposable Subscribe(IObserver<int> observer)
        {
            return Observable.Return(1).Asynchronous().Subscribe(observer);
        }

        #endregion

        #region Implementation of IDisposable

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        public void Dispose()
        {
            OnCompleted();
        }

        #endregion
    }
}