Edit Question - KnowledgeHub

Title Body

The Reactive Extensions come with a lot of helper methods for turning existing events and asynchronous operations into observables but how would you implement an IObservable<T> from scratch? IEnumerable has the lovely yield keyword to make it very simple to implement. What is the proper way of implementing IObservable<T>? Do I need to worry about thread safety? I know there is support for getting called back on a specific synchronization context but is this something I as an IObservable<T> author need to worry about or this somehow built-in? update: Here's my C# version of Brian's F# solution using System; using System.Linq; using Microsoft.FSharp.Collections; namespace Jesperll { class Observable<T> : IObservable<T>, IDisposable where T : EventArgs { private FSharpMap<int, IObserver<T>> subscribers = FSharpMap<int, IObserver<T>>.Empty; 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 && !isDisposed) { 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 (exce

Tags (comma-separated) Cancel