ObserverCollection.cs @@ -4,10 +4,10 @@ namespace Reactive { public static class ButtonExtensions { - public static IObservable<MouseEventArgs> GetMouseDowns(this Button b) + public static IObservable<MouseEventArgs> GetMouseDowns(this Button button) { var wrapper = new EventWrapper<MouseEventArgs>(); - b.MouseDown += wrapper.Handle; + button.MouseDown += wrapper.Handler; return wrapper; } } ButtonExtensions.cs @@ -1,75 +1,15 @@ using System; -using System.Collections.Generic; using System.Windows.Forms; namespace Reactive { - public class ObserverCollection<T> : IObservable<T> - { - protected IList<IObserver<T>> _observers; - - public ObserverCollection() - { - _observers = new List<IObserver<T>>(); - } - - public virtual IDisposable Subscribe(IObserver<T> observer) - { - _observers.Add(observer); - return new Disposer(() => Remove(observer)); - } - - public void NotifyNext(T t) - { - foreach (IObserver<T> observer in _observers) - observer.OnNext(t); - } - - public void NotifyDone() - { - foreach (IObserver<T> observer in _observers) - observer.OnDone(); - } - - public void NotifyError(Exception e) - { - foreach (IObserver<T> observer in _observers) - observer.OnError(e); - } - - protected virtual void Remove(IObserver<T> observer) - { - _observers.Remove(observer); - } - - public void Clear() - { - _observers.Clear(); - } - - protected class Disposer : IDisposable - { - private readonly Action _disposeAction; - - public Disposer(Action disposeAction) - { - _disposeAction = disposeAction; - } - - public void Dispose() - { - _disposeAction(); - } - } - } - public class EventWrapper<T> : IObservable<T> where T : EventArgs { private readonly ObserverCollection<T> _observers = new ObserverCollection<T>(); - public void Handle(object sender, T t) + public void Handler(object sender, T eventArgs) { - _observers.NotifyNext(t); + _observers.NotifyNext(eventArgs); } public IDisposable Subscribe(IObserver<T> observer) EventWrapper.cs @@ -6,8 +6,7 @@ namespace Reactive { public static IObservable<TResult> Select<T, TResult>(this IObservable<T> observable, Func<T, TResult> func) { - return new SelectObservable<T, TResult>(observable, func); - //return ObservableBuilder.Create((IObserver<TResult> observer) => observable.Subscribe(ObserverBuilder.Create(observer, (T a) => observer.OnNext(func(a))))); + return new SelectObservable<T, TResult>(observable, func); } public static IObservable<TResult> SelectMany<TSource, TCollection, TResult>(this IObservable<TSource> source, Func<TSource, IObservable<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector) Extensions.cs @@ -15,18 +15,23 @@ namespace Reactive { InitializeComponent(); - Func<MouseEventArgs, int> slowOperation = args => - { - System.Threading.Thread.Sleep(3000); - if(args.Button == MouseButtons.Middle) - throw new Exception("MIDDLE BUTTON NOT ALLOWED!!!"); - return args.X; - }; + //Func<MouseEventArgs, int> slowOperation = args => + //{ + // System.Threading.Thread.Sleep(3000); + // if(args.Button == MouseButtons.Middle) + // throw new Exception("MIDDLE BUTTON NOT ALLOWED!!!"); + // return args.X; + //}; + + //IObservable<string> messages = from md in button1.GetMouseDowns() + // from x in slowOperation.AsAsyncObservable(md) + // where md.Button == MouseButtons.Right + // select "Mouse down: " + x + "\n"; + + IObservable<MouseEventArgs> mouseEvents = button1.GetMouseDowns(); - IObservable<string> messages = from md in button1.GetMouseDowns() - from x in slowOperation.AsAsyncObservable(md) - where md.Button == MouseButtons.Right - select "Mouse down: " + x + "\n"; + IObservable<string> messages = from md in mouseEvents + select "Mouse down at: " + md.X + "\n"; messages.Subscribe(new TextBoxUpdater(textBox1)); } @@ -40,7 +45,7 @@ namespace Reactive _textBox = textBox; } - private void SetText(string text) + private void AppendText(string text) { Action textboxUpdater = () => _textBox.AppendText(text); _textBox.BeginInvoke(textboxUpdater); @@ -48,17 +53,17 @@ namespace Reactive public void OnNext(string s) { - SetText(s); + AppendText(s); } public void OnDone() { - SetText("Done\n"); + AppendText("Done\n"); } public void OnError(Exception e) { - SetText("Error: " + e.Message); + AppendText("Error: " + e.Message); } } Form1.cs @@ -9,7 +9,7 @@ namespace Reactive public interface IObserver<T> { - void OnNext(T t); + void OnNext(T item); void OnDone(); void OnError(Exception e); } @@ -23,9 +23,9 @@ namespace Reactive _action = action; } - public void OnNext(T t) + public void OnNext(T item) { - _action(t); + _action(item); } IObservable.cs @@ -20,9 +20,9 @@ namespace Reactive _onNext = onNext; } - public void OnNext(T2 t) + public void OnNext(T2 item) { - _onNext(t); + _onNext(item); } public void OnDone() ObserverBuilder.cs @@ -61,6 +61,7 @@ </Compile> <Compile Include="IObservable.cs" /> <Compile Include="ObserverBuilder.cs" /> + <Compile Include="ObserverCollection.cs" /> <Compile Include="SelectManyObservable.cs" /> <Compile Include="SelectObservable.cs" /> <Compile Include="Program.cs" /> Reactive.csproj @@ -19,27 +19,4 @@ namespace Reactive } } - //public class ObservableBuilder - //{ - // public static IObservable<T> Create<T>(Func<IObserver<T>, IDisposable> subscribe) - // { - // return new ObservableWrapper<T>(subscribe); - // } - - // private class ObservableWrapper<T> : IObservable<T> - // { - // private readonly Func<IObserver<T>, IDisposable> _subscribe; - - // public ObservableWrapper(Func<IObserver<T>, IDisposable> subscribe) - // { - // _subscribe = subscribe; - // } - - // public IDisposable Subscribe(IObserver<T> observer) - // { - // return _subscribe(observer); - // } - // } - //} - } \ No newline at end of file SelectObservable.cs c6e516ae76429bcd8b610bca6b0ffa67aa131e86 Paul Batum paul.batum@gmail.com http://github.com/paulbatum/Reactive/commit/a655578cf6981a73f7b9ecc5d2380f98824f97ec a655578cf6981a73f7b9ecc5d2380f98824f97ec 2009-07-19T02:20:19-07:00 2009-07-19T02:20:19-07:00 Cleaned up dead code, made minor readability improvements. e15ee93b4a120c38706deae02bdc878716d65eba Paul Batum paul.batum@gmail.com