Observable

class Observable<T>:IObservable<T>,IDisposable {
            private Dictionary<int,IObserver<T>> subscribers = new Dictionary<int,IObserver<T>>();
            private readonly object thisLock = new object();
            private int key;
            private bool isDisposed;

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

            protected virtual void Dispose(bool disposing) {
                if(disposing) {
                    OnCompleted();
                    isDisposed = true;
                }
            }

            protected void OnNext(T value) {
                if(isDisposed) {
                    throw new ObjectDisposedException("Observable<T>");
                }

                foreach(IObserver<T> observer in subscribers.Select(kv => kv.Value)) {
                    observer.OnNext(value);
                }
            }

            protected void OnError(Exception exception) {
                if(isDisposed) {
                    throw new ObjectDisposedException("Observable<T>");
                }

                if(exception == null) {
                    throw new ArgumentNullException("exception");
                }

                foreach(IObserver<T> observer in subscribers.Select(kv => kv.Value)) {
                    observer.OnError(exception);
                }
            }

            protected void OnCompleted() {
                if(isDisposed) {
                    throw new ObjectDisposedException("Observable<T>");
                }

                foreach(IObserver<T> observer in subscribers.Select(kv => kv.Value)) {
                    observer.OnCompleted();
                }
            }

            public IDisposable Subscribe(IObserver<T> observer) {
                if(observer == null) {
                    throw new ArgumentNullException("observer");
                }

                lock(thisLock) {
                    int k = key++;
                    subscribers.Add(k,observer);
                    return new AnonymousDisposable(() => {
                        lock(thisLock) {
                            subscribers.Remove(k);
                        }
                    });
                }
            }
        }

        class AnonymousDisposable:IDisposable {
            Action dispose;
            public AnonymousDisposable(Action dispose) {
                this.dispose = dispose;
            }

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

        public class AnonymousObservable<T>:IObservable<T> {
            private Func<IObserver<T>,IDisposable> _subscribe;
            public AnonymousObservable(Func<IObserver<T>,IDisposable> subscribe) {
                _subscribe = subscribe;
            }

            public IDisposable Subscribe(IObserver<T> observer) {
                return _subscribe(observer);
            }
        }