Implementing Observers and Subjects using IObserver/IObservable

I want to create a class that can be used to represent a dynamically computed value, and another class that represents a value can be the source (subject) for these dynamically computed values. The goal is that when the subject changes, the computed value is updated automatically.

It seems to me that using IObservable/IObserver is the way to go. Unfortunately I can't use the Reactive Extensions library, so I am forced to implement the subject/observer pattern from scratch.

Enough blabla, here are my classes:

public class Notifier<T> : IObservable<T>
{
    public Notifier();
    public IDisposable Subscribe(IObserver<T> observer);
    public void Subscribe(Action<T> action);
    public void Notify(T subject);
    public void EndTransmission();
}

public class Observer<T> : IObserver<T>, IDisposable
{
    public Observer(Action<T> action);
 开发者_JAVA技巧   public void Subscribe(Notifier<T> tracker);
    public void Unsubscribe();
    public void OnCompleted();
    public void OnError(Exception error);
    public void OnNext(T value);
    public void Dispose();
}

public class ObservableValue<T> : Notifier<T>
{
    public T Get();
    public void Set(T x);
}

public class ComputedValue<T>
{
    public T Get();
    public void Set(T x);
}

My implementation is lifted mostly from: http://msdn.microsoft.com/en-us/library/dd990377.aspx.

So what would the "right" way to do this be? Note: I don't care about LINQ or multi-threading or even performance. I just want it to be simple and easy to understand.

---

If I were you I would try to implement your classes as closely as possible to the way Rx has been implemented.

One of the key underlying principles is the use of relatively few concrete classes that are combined using a large number of operations. So you should create a few basic building blocks and use composition to bring them all together.

There are two classes I would take an initial look at under Reflector.NET: AnonymousObservable<T> & AnonymousObserver<T>. In particular AnonymousObservable<T> is used through-out Rx as the basis for instantiating observables. In fact, if you look at the objects that derive from IObservable<T> there are a few specialized implementations, but only AnonymousObservable<T> is for general purpose use.

The static method Observable.Create<T>() is essentially a wrapper to AnonymousObservable<T>.

The other Rx class that is clearly a fit for your requirements is BehaviorSubject<T>. Subjects are both observables and observers and BehaviorSubject fits your situation because it remembers the last value that is received.

Given these basic classes then you almost have all of the bits you need to create your specific objects. Your objects shouldn't inherit from the above code, but instead use composition to bring together the behaviour that you need.

Now, I would suggest some changes to your class designs to make them more compatible with Rx and thus more composible and robust.

I would drop your Notifier<T> class in favour of using BehaviourSubject<T>.

I would drop your Observer<T> class in favour of using AnonymousObserver<T>.

Then I would modify ObservableValue<T> to look like this:

public class ObservableValue<T> : IObservable<T>, IDisposable
{
    public ObservableValue(T initial) { ... }
    public T Value { get; set; }
    public IDisposable Subscribe(IObserver<T> observer);
    public void Dispose();
}

The implementation of ObservableValue<T> would wrap BehaviourSubject<T> rather than inherit from it as exposing the IObserver<T> members would allow access to OnCompleted & OnError which wouldn't make too much sense since this class represents a value and not a computation. Subscriptions would use AnonymousObservable<T> and Dispose would clean up the wrapped BehaviourSubject<T>.

Then I would modify ComputedValue<T> to look like this:

public class ComputedValue<T> : IObservable<T>, IDisposable
{
    public ComputedValue(IObservable<T> source) { ... }
    public T Value { get; }
    public IDisposable Subscribe(IObserver<T> observer);
    public void Dispose();
}

The ComputedValue<T> class would wrap AnonymousObservable<T> for all subscribers and and use source to grab a local copy of the values for the Value property. The Dispose method would be used to unsubscribe from the source observable.

These last two classes are the only real specific implementation your design appears to need - and that's only because of the Value property.

Next you need a static ObservableValues class for your extension methods:

public static class ObservableValues
{
    public static ObservableValue<T> Create<T>(T initial)
    { ... }

    public static ComputedValue<V> Compute<T, U, V>(
        this IObservable<T> left,
        IObservable<U> right,
        Func<T, U, V> computation)
    { ... }
}

The Compute method would use an AnonymousObservable<V> to perform the computation and produce an IObservable<V> to pass to the constructor of ComputedValue<V> that is returned by the method.

With all this in place you can now write this code:

var ov1 = ObservableValues.Create(1);
var ov2 = ObservableValues.Create(2);
var ov3 = ObservableValues.Create(3);

var cv1 = ov1.Compute(ov2, (x, y) => x + y);
var cv2 = ov3.Compute(cv1, (x, y) => x * y);

//cv2.Value == 9

ov1.Value = 2;
ov2.Value = 3;
ov3.Value = 4;

//cv2.Value == 20

Please let me know if this is helpful and/or if there is anything that I can elaborate on.

---

EDIT: Also need some disposables.

You'll also need to implement AnonymousDisposable & CompositeDisposable to manage your subscriptions particularly in the Compute extension method. Take a look at the Rx implementations using Reflector.NET or use my versions below.

public sealed class AnonymousDisposable : IDisposable
{
    private readonly Action _action;
    private int _disposed;

    public AnonymousDisposable(Action action)
    {
        _action = action;
    }

    public void Dispose()
    {
        if (Interlocked.Exchange(ref _disposed, 1) == 0)
        {
            _action();
        }
    }
}

public sealed class CompositeDisposable : IEnumerable<IDisposable>, IDisposable
{
    private readonly List<IDisposable> _disposables;
    private bool _disposed;

    public CompositeDisposable()
        : this(new IDisposable[] { })
    { }

    public CompositeDisposable(IEnumerable<IDisposable> disposables)
    {
        if (disposables == null) { throw new ArgumentNullException("disposables"); }
        this._disposables = new List<IDisposable>(disposables);
    }

    public CompositeDisposable(params IDisposable[] disposables)
    {
        if (disposables == null) { throw new ArgumentNullException("disposables"); }
        this._disposables = new List<IDisposable>(disposables);
    }

    public void Add(IDisposable disposable)
    {
        if (disposable == null) { throw new ArgumentNullException("disposable"); }
        lock (_disposables)
        {
            if (_disposed)
            {
                disposable.Dispose();
            }
            else
            {
                _disposables.Add(disposable);
            }
        }
    }

    public IDisposable Add(Action action)
    {
        if (action == null) { throw new ArgumentNullException("action"); }
        var disposable = new AnonymousDisposable(action);
        this.Add(disposable);
        return disposable;
    }

    public IDisposable Add<TDelegate>(Action<TDelegate> add, Action<TDelegate> remove, TDelegate handler)
    {
        if (add == null) { throw new ArgumentNullException("add"); }
        if (remove == null) { throw new ArgumentNullException("remove"); }
        if (handler == null) { throw new ArgumentNullException("handler"); }
        add(handler);
        return this.Add(() => remove(handler));
    }

    public void Clear()
    {
        lock (_disposables)
        {
            var disposables = _disposables.ToArray();
            _disposables.Clear();
            Array.ForEach(disposables, d => d.Dispose());
        }
    }

    public void Dispose()
    {
        lock (_disposables)
        {
            if (!_disposed)
            {
                this.Clear();
            }
            _disposed = true;
        }
    }

    public IEnumerator<IDisposable> GetEnumerator()
    {
        lock (_disposables)
        {
            return _disposables.ToArray().AsEnumerable().GetEnumerator();
        }
    }

    IEnumerator IEnumerable.GetEnumerator()
    {
        return this.GetEnumerator();
    }

    public bool IsDisposed
    {
        get
        {
            return _disposed;
        }
    }
}