Showing off the Observable Pattern

AnonymousObserver.cs

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

namespace ConsoleApplication2
{
  public sealed class AnonymousObserver<T> : IObserver<T>
  {
    private readonly Action<T> _onNext;
    private readonly Action<Exception> _onError;
    private readonly Action _onCompleted;

    public AnonymousObserver( Action<T> onNext, Action<Exception> onError, Action onCompleted )
    {      
      if ( onNext == null )
        throw new ArgumentNullException( nameof( onNext ) );
      if ( onError == null )
        throw new ArgumentNullException( nameof( onError ) );
      if ( onCompleted == null )
        throw new ArgumentNullException( nameof( onCompleted ) );
      _onNext = onNext;
      _onError = onError;
      _onCompleted = onCompleted;
    }

    public void OnNext( T value )
    {
      _onNext( value );
    }

    public void OnError( Exception error )
    {
      if ( error == null )
        throw new ArgumentNullException( nameof( error ) );
      _onError(error);
    }

    public void OnCompleted()
    {
      _onCompleted();
    }
  }
}

Disposable.cs

using System;

namespace ConsoleApplication2
{
  public class Disposable : IDisposable
  {
    private readonly Action _action;

    public static IDisposable Create( Action action)
    {
      return new Disposable( action );
    }

    public Disposable( Action action )
    {
      if ( action == null )
        throw new ArgumentNullException( nameof( action ) );
      _action = action;
    }

    public void Dispose()
    {
      _action();
    }
  }
}

Observable.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace ConsoleApplication2
{
  public class Observable<T> : IObservable<T>
  {
    private readonly HashSet<IObserver<T>> _observers = new HashSet<IObserver<T>>();

    public IDisposable Subscribe( IObserver<T> observer )
    {
      _observers.Add( observer );
      return Disposable.Create( () => _observers.Remove( observer ));
    }

    public IDisposable Subscribe( Action<T> onNext, Action<Exception> onError, Action onCompleted )
    {
      return Subscribe(new AnonymousObserver<T>( onNext, onError, onCompleted ));
    }

    public void Notify( T message)
    {
      foreach ( var observer in _observers )
        observer.OnNext( message );
    }
  }
}

Program.cs

using System;
using System.Threading;

namespace ConsoleApplication2
{
  class Program
  {
    static void Main()
    {
      var stateMessanger = new Observable<StateMessage>();
      var observer = stateMessanger.Subscribe( PrintMessage, ErrorOccured, OnCompleted );

      // Send State Change to Observers
      stateMessanger.Notify( CreateMessage() );
      Thread.Sleep( 1000 );
      stateMessanger.Notify( CreateMessage() );
      Thread.Sleep( 1000 );
      Console.WriteLine( "cleaning up observer" );
      observer.Dispose();
      stateMessanger.Notify( CreateMessage() );
      Console.WriteLine( "message will not be received because no one is watching it now" );
    }

    public static StateMessage CreateMessage()
    {
      return new StateMessage( "StateChanged at " + DateTime.Now );
    }

    public static void PrintMessage( StateMessage message)
    {
      Console.WriteLine( "Message Received  {0}", message );
    }

    public static void ErrorOccured( Exception e)
    {
      Console.WriteLine( "Exception Occured in StateServer: {0} : {1}", e, DateTime.Now   );
    }

    public static void OnCompleted()
    {
      Console.WriteLine( "No More messsage {0}", DateTime.Now);
    }
  }
}

StateMessage.cs

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

namespace ConsoleApplication2
{
  public class StateMessage
  {
    public string Message { get; private set; }

    public StateMessage( string message )
    {
      Message = message;
    }

    public override string ToString()
    {
      return Message;
    }
  }
}