SourceForge.net

 using System;

 using System.Threading;

 using System.Collections.Generic;

 using System.Linq;

 using System.Text;

 using System.Reactive;

 using System.Reactive.Linq;

 namespace Sasa.Reactive

 {

     /// <summary>

     /// Indicates the thread should retry its updates.

     /// </summary>

     public sealed class RetryException : System.Exception

     {

     }

     /// <summary>

     /// Indicates an error occurred while propagating the update.

     /// </summary>

     public sealed class UpdateException : System.Exception

     {

         /// <summary>

         /// Construct an exception.

         /// </summary>

         /// <param name="message">The message to report.</param>

         /// <param name="inner">The inner exception.</param>

         public UpdateException(string message, Exception inner)

             : base(message, inner)

         {

         }

     }

     /// <summary>

     /// Used only

     /// </summary>

     public sealed class NoValueException : System.Exception

     {

         internal static NoValueException instance = new NoValueException();

         NoValueException()

             : base("The Property<T> currently has no value.")

         {

         }

     }

     /// <summary>

     /// An observable property that retains only the current value, and notifies clients only of distinct changes.

     /// </summary>

     /// <typeparam name="T">The type of values.</typeparam>

     /// <remarks>

     /// Clients should first invoke <seealso cref="Ticket.Acquire" /> or <seealso cref="Transaction.Begin" /> before operating on signals.

     /// </remarks>

     public class Property<T> : IObservable<T>, IObserver<T>, IDisposable, IRef<T>, IValue<T>, IResult<T>

     {

         // the queue of subscribed observers

         internal Node head;

         internal Node tail;

         // the current value or error

         internal T value;

         internal Exception error;

         // monotonically increasing version number; crude approximation of transactional memory

         long version = long.MinValue;

         // the current subscription

         IDisposable subscription;

         #region Public API

         /// <summary>

         /// Initialize an empty signal.

         /// </summary>

         public Property() : this(NoValueException.instance)

         {

         }

         /// <summary>

         /// Initialize a signal with a default value.

         /// </summary>

         /// <param name="value">The default value of the signal.</param>

         public Property(T value)

             : this(error: null)

         {

             this.value = value;

         }

         /// <summary>

         /// Initialize a signal with a default value.

         /// </summary>

         /// <param name="error">The default value of the signal.</param>

         public Property(Exception error)

         {

             head = tail = new Node();

             this.error = error;

         }

         /// <summary>

         /// Initialize the signal with the given observable sequence.

         /// </summary>

         /// <param name="observable">The observable to listen to.</param>

         public Property(IObservable<T> observable)

             : this(observable.FirstOrDefault())

         {

             Switch(observable);

         }

         /// <summary>

         /// Dispose of a <seealso cref="Property{T}"/>.

         /// </summary>

         ~Property()

         {

             Dispose();

         }

         /// <summary>

         /// The current value of the signal.

         /// </summary>

         /// <exception cref="System.Exception">Thrown if the current value is an error.</exception>

         public T Value

         {

             get

             {

                 //FIXME: why lock this?

                 lock (head)

                 {

                     Acquire();

                     if (error != null) throw error;

                     return value;

                 }

             }

             set

             {

                 bool changed;

                 //FIXME: why lock this?

                 lock (head)

                 {

                     Acquire();

                     changed = !EqualityComparer<T>.Default.Equals(this.value, this.value = value);

                     this.error = null;

                 }

                 if (changed) Notify(head.next, value, null);

             }

         }

         /// <summary>

         /// The current error value of the signal, if any.

         /// </summary>

         public Exception Error

         {

             get

             {

                 lock (head)

                 {

                     Acquire();

                     return error;

                 }

             }

             set

             {

                 T ovalue;

                 bool changed;

                 lock (head)

                 {

                     Acquire();

                     changed = !EqualityComparer<Exception>.Default.Equals(this.error, this.error = value);

                     ovalue = this.value;

                 }

                 if (changed) Notify(head.next, ovalue, value);

             }

         }

         /// <summary>

         /// True if the <seealso cref="Property{T}"/> has a value, and no error.

         /// </summary>

         public bool HasValue

         {

             get { return Error != null; }

         }

         /// <summary>

         /// Obtains the encapsulated value, if any.

         /// </summary>

         /// <param name="value">The value to extract.</param>

         /// <returns>True if value was available, false otherwise.</returns>

         public bool TryGetValue(out T value)

         {

             value = this.value;

             return this.error == null;

         }

         /// <summary>

         /// Returns true if this signal has subscribed to observables.

         /// </summary>

         public bool IsSubscribed

         {

             get { return subscription != null; }

         }

         /// <summary>

         /// Notify <paramref name="observer"/> of any changes to this signal.

         /// </summary>

         /// <param name="observer">The observer to notify.</param>

         /// <returns>A disposable handle representing the subscription.</returns>

         public IDisposable Subscribe(IObserver<T> observer)

         {

             // notify current observer

             T value;

             Exception error;

             //FIXME: why acquire ticket?

             Ticket.Acquire();

             lock (head)

             {

                 Acquire();

                 value = this.value;

                 error = this.error;

             }

             var node = Enqueue(observer);

             // does not notify if property has not yet been initialized

             Notify(node, value, error);

             // if ticket acquired locally, then release it immediately after notifications complete

             Ticket.Release();

             return new DisposeObserver { node = node, property = this };

         }

         /// <summary>

         /// Subscribe to any changes made to this property.

         /// </summary>

         /// <param name="onNext">The delegate to invoke for a new value.</param>

         /// <returns>A disposable handle representing the subscription.</returns>

         public IDisposable Subscribe(Action<T> onNext)

         {

             return Subscribe(Observer.Create(onNext));

         }

         /// <summary>

         /// Subscribe to any changes made to this property.

         /// </summary>

         /// <param name="onNext">The delegate to invoke for a new value.</param>

         /// <param name="onError">The delegate to invoke when the property receives an error.</param>

         /// <returns>A disposable handle representing the subscription.</returns>

         public IDisposable Subscribe(Action<T> onNext, Action<Exception> onError)

         {

             return Subscribe(Observer.Create(onNext, onError));

         }

         /// <summary>

         /// Subscribe to any changes made to this property.

         /// </summary>

         /// <param name="onNext">The delegate to invoke for a new value.</param>

         /// <param name="onError">The delegate to invoke on error.</param>

         /// <param name="onDispose">The delegate to invoke when the property is disposed.</param>

         /// <returns>A disposable handle representing the subscription.</returns>

         public IDisposable Subscribe(Action<T> onNext, Action<Exception> onError, Action onDispose)

         {

             return Subscribe(Observer.Create(onNext, onError, onDispose));

         }

         /// <summary>

         /// Subscribe this signal to listen to <paramref name="observable"/>.

         /// </summary>

         /// <param name="observable">The observable to listen to.</param>

         public void Switch(IObservable<T> observable)

         {

             //FIXME: why acquire this lock?

             lock (head)

             {

                 Acquire();

                 if (subscription != null) subscription.Dispose();

                 subscription = observable.Subscribe(this);

             }

         }

         /// <summary>

         /// Dispose of any subscriptions.

         /// </summary>

         public void Dispose()

         {

             Node head;

             lock (this.head)

             {

                 tail = null;

                 if (subscription != null) subscription.Dispose();

                 head = this.head.next;

                 this.head.next = null;

             }

             for (var y = head; y != null; y = y.next)

             {

                 try

                 {

                     y.observer.OnCompleted();

                 }

                 catch

                 {

                     continue;

                 }

             }

         }

         /// <summary>

         /// Generates a string representation.

         /// </summary>

         /// <returns>A string representation of this property.</returns>

         public override string ToString()

         {

             return Value.ToString();

         }

         #endregion

         #region Private API

         void IObserver<T>.OnNext(T value)

         {

             Value = value;

         }

         void IObserver<T>.OnError(Exception error)

         {

             Error = error;

         }

         void IObserver<T>.OnCompleted()

         {

             Dispose();

         }

         void Acquire()

         {

             CheckDispose();

             if (version > Ticket.Current) throw new RetryException();

             version = Ticket.Current;

         }

         void CheckDispose()

         {

             if (tail == null) throw new ObjectDisposedException("Property<T>", "This property has been disposed.");

         }

         void Notify(Node head, T value, Exception error)

         {

             if (error != null)

             {

                 if (error is NoValueException) return;

                 for (var x = head; x != null; x = x.next)

                 {

                     Notify(x.observer, error);

                 }

             }

             else

             {

                 for (var x = head; x != null; x = x.next)

                 {

                     Notify(x.observer, value);

                 }

             }

         }

         void Notify(IObserver<T> x, T value)

         {

             try

             {

                 x.OnNext(value);

             }

             catch (RetryException)

             {

                 //FIXME: should this propagate, ignore, or accumulate errors?

                 throw; // propagate retry exceptions

             }

             catch (Exception e)

             {

                 // notify only the current observer of the error

                 Notify(x, new UpdateException("Error propagating value", e));

             }

         }

         void Notify(IObserver<T> x, Exception error)

         {

             while (true)

             {

                 try

                 {

                     x.OnError(error);

                     return;

                 }

                 catch (RetryException)

                 {

                     //FIXME: should this propagate, ignore, or accumulate errors?

                     throw; // propagate retry exceptions

                 }

                 catch (Exception e)

                 {

                     // try notifying of new update error

                     error = new UpdateException("Error propagating exception.", e);

                 }

             }

         }

         internal Node Enqueue(IObserver<T> o)

         {

             lock (tail)

             {

                 tail.next = new Node { observer = o };

                 return tail = tail.next;

             }

         }

         void Remove(Node node)

         {

             // must lock for thread-safety:

             // head -> o1 -> o2 -> o3

             //T1:rem(o1)       T2:rem(o2)

             //head.n = o1.n    o1.n = o2.n

             lock (head)

             {

                 for (var x = head; x.next != null; x = x.next)

                 {

                     if (node == x.next)

                     {

                         // must lock in case node is tail

                         lock (tail) x.next = node.next;

                         return;

                     }

                 }

             }

         }

         #endregion

         #region Public operators

         /// <summary>

         /// Implicitly create a <seealso cref="Property{T}"/>.

         /// </summary>

         /// <param name="value">The default value of the property.</param>

         /// <returns>A property with the given value.</returns>

         public static implicit operator Property<T>(T value)

         {

             return new Property<T>(value);

         }

         /// <summary>

         /// Implicitly create a <seealso cref="Property{T}"/>.

         /// </summary>

         /// <param name="error">The starting error of the property.</param>

         /// <returns>A property with the given error.</returns>

         public static implicit operator Property<T>(Exception error)

         {

             return new Property<T>(error);

         }

         //public static implicit operator Signal<T>(IObservable<T> value)

         //{

         //    return new Signal<T>(value);

         //}

         /// <summary>

         /// Merge events from two sources.

         /// </summary>

         /// <param name="lhs">The first property.</param>

         /// <param name="rhs">The second property.</param>

         /// <returns>A stream accumulating the changes from both sources.</returns>

         public static IObservable<T> operator |(Property<T> lhs, IObservable<T> rhs)

         {

             return lhs.Merge(rhs);

         }

         /// <summary>

         /// Merge events from two properties.

         /// </summary>

         /// <param name="lhs">The first property.</param>

         /// <param name="rhs">The second property.</param>

         /// <returns>A stream accumulating the changes from both sources.</returns>

         public static IObservable<T> operator |(IObservable<T> lhs, Property<T> rhs)

         {

             return lhs.Merge(rhs);

         }

         /// <summary>

         /// Merge events from two properties.

         /// </summary>

         /// <param name="lhs">The first property.</param>

         /// <param name="rhs">The second property.</param>

         /// <returns>A stream accumulating the changes from both sources.</returns>

         public static IObservable<T> operator |(Property<T> lhs, Property<T> rhs)

         {

             return lhs.Merge(rhs);

         }

         #endregion

         class DisposeObserver : IDisposable

         {

             internal Node node;

             internal Property<T> property;

             ~DisposeObserver()

             {

                 Dispose();

             }

             public void Dispose()

             {

                 var x = Interlocked.CompareExchange(ref property, null, property);

                 if (x != null)

                 {

                     x.Remove(node);

                     GC.SuppressFinalize(this);

                 }

             }

         }

         internal sealed class Node

         {

             internal IObserver<T> observer;

             internal Node next;

         }

         #region Linq extensions

         /// <summary>

         /// 

         /// </summary>

         /// <typeparam name="R"></typeparam>

         /// <param name="selector"></param>

         /// <returns></returns>

         public Property<R> Select<R>(Func<T, R> selector)

         {

             var x = new Property<R>();

                 x.subscription = Subscribe(t =>

                 {

                     try

                     {

                         x.Value = selector(t);

                     }

                     catch (Exception e)

                     {

                         x.Error = e;

                     }

                 },

                 e => x.Error = e);

             return x;

         }

         public Property<R> SelectMany<R>(Func<T, Property<R>> selector)

         {

             var x = new Property<R>();

                 x.subscription = Subscribe(t =>

                 {

                     try

                     {

                         x.Value = selector(t).Value;

                     }

                     catch (Exception e)

                     {

                         x.Error = e;

                     }

                 }, e => x.Error = e);

             return x;

         }

         public Property<R> SelectMany<U, R>(Func<T, Property<U>> collector, Func<T, U, R> selector)

         {

             var x = new Property<R>();

                 x.subscription = Subscribe(t =>

                 {

                     try

                     {

                         x.Value = selector(t, collector(t).Value);

                     }

                     catch (Exception e)

                     {

                         x.Error = e;

                     }

                 }, e => x.Error = e);

             return x;

         }

         public Property<T> Where(Func<T, bool> predicate)

         {

             var x = new Property<T>();

                 x.subscription = Subscribe(t =>

                 {

                     try

                     {

                         if (predicate(t)) x.Value = t;

                     }

                     catch (Exception e)

                     {

                         x.Error = e;

                     }

                 },

                 e => x.Error = e);

             return x;

         }

         public Property<T> Skip(int count)

         {

             var x = new Property<T>();

                 x.subscription = Subscribe(t =>

                 {

                     if (count-- > 0) return;

                     try

                     {

                         x.Value = t;

                     }

                     catch (Exception e)

                     {

                         x.Error = e;

                     }

                 },

                 e => x.Error = e);

             return x;

         }

         /// <summary>

         /// Register a permanent event handler.

         /// </summary>

         /// <param name="onNext">The handler to invoke.</param>

         public void Register(Action<T> onNext)

         {

             Enqueue(Observer.Create(onNext));

         }

         #endregion

     }

     /// <summary>

     /// Some convenient extensions.

     /// </summary>

     public static class Property

     {

         /// <summary>

         /// Bind an object property.

         /// </summary>

         /// <typeparam name="T">The type of the property value.</typeparam>

         /// <param name="value">The current property value.</param>

         /// <param name="setter">A function used to set the underlying property value.</param>

         /// <returns>A property that's bidirectionally bound to the underlying property.</returns>

         /// <remarks>

         /// The returned Property<T> must be the ONLY way that the underlying object property is updated,

         /// otherwise they will become out of sync. Use the more elaborate Bind operation if the underlying

         /// property is updated by other means.

         /// </remarks>

         public static Property<T> Bind<T>(this T value, Action<T> setter)

         {

             var prop = new Property<T>(value);

                 prop.Enqueue(Observer.Create(setter));

             return prop;

         }

         /// <summary>

         /// Bind an object property.

         /// </summary>

         /// <typeparam name="T">The type of the property value.</typeparam>

         /// <typeparam name="TEventArgs">The event args.</typeparam>

         /// <param name="getter">A function used to obtain the underlying property value.</param>

         /// <param name="setter">A function used to set the underlying property value.</param>

         /// <param name="register">Registers an event handler that notifies when the underlying property has been updated.</param>

         /// <param name="unregister">Unregisters an event handler for the underlying property.</param>

         /// <returns>A property that's bidirectionally bound to an underlying object property.</returns>

         public static Property<T> Bind<T, TEventArgs>(Func<T> getter, Action<T> setter, Action<EventHandler<TEventArgs>> register, Action<EventHandler<TEventArgs>> unregister = null)

             where TEventArgs : EventArgs

         {

             var prop = new Property<T>();

             EventHandler<TEventArgs> h = (o, e) =>

             {

                 try

                 {

                     prop.Value = getter();

                 }

                 catch (Exception ex)

                 {

                     prop.Error = ex;

                 }

             };

             // invoke event handler to set the initial property

             h(prop, default(TEventArgs));

             prop.Enqueue(Observer.Create(setter, () => unregister(h)));

             register(h);

             return prop;

         }

     }

 }