Implementando IObservable <T> desde cero

Question

Las extensiones reactivas vienen con una gran cantidad de métodos auxiliares para convertir eventos existentes y operaciones asincrónicas en observables, pero ¿cómo implementaría un IObservable <T> desde cero?

IEnumerable tiene la encantadora palabra clave yield para que sea muy simple de implementar.

¿Cuál es la forma correcta de implementar IObservable <T>?

¿Debo preocuparme por la seguridad de los hilos?

Sé que hay soporte para recibir una llamada de vuelta en un contexto de sincronización específico, pero ¿es esto algo que yo como autor IObservable <T> debo preocuparme o esto de alguna manera está incorporado?

actualización:

Aquí está mi C# versión de la solución de Brian F #

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 (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 = subscribers.Add(k, observer); 
       return new AnonymousDisposable(() => 
       { 
        lock (thisLock) 
        { 
         subscribers = subscribers.Remove(k); 
        } 
       }); 
      } 
     } 
    } 

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

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

edición: No tire ObjectDisposedException si Desechar se llama dos veces

Answer 1

Honestamente, no estoy seguro de qué tan "correcto" es todo esto, pero si se siente bastante bien según mi experiencia hasta ahora. Es código F #, pero con suerte obtienes una idea del sabor. Le permite 'actualizar' un objeto de origen, al que puede llamar Siguiente/Completado/Error, y administra las suscripciones e intenta Assert cuando el origen o los clientes hacen cosas malas.

type ObservableSource<'T>() =  // ' 
    let protect f = 
     let mutable ok = false 
     try 
      f() 
      ok <- true 
     finally 
      Debug.Assert(ok, "IObserver methods must not throw!") 
      // TODO crash? 
    let mutable key = 0 
    // Why a Map and not a Dictionary? Someone's OnNext() may unsubscribe, so we need threadsafe 'snapshots' of subscribers to Seq.iter over 
    let mutable subscriptions = Map.empty : Map<int,IObserver<'T>> // ' 
    let next(x) = subscriptions |> Seq.iter (fun (KeyValue(_,v)) -> protect (fun() -> v.OnNext(x))) 
    let completed() = subscriptions |> Seq.iter (fun (KeyValue(_,v)) -> protect (fun() -> v.OnCompleted())) 
    let error(e) = subscriptions |> Seq.iter (fun (KeyValue(_,v)) -> protect (fun() -> v.OnError(e))) 
    let thisLock = new obj() 
    let obs = 
     { new IObservable<'T> with  // ' 
      member this.Subscribe(o) = 
       let k = 
        lock thisLock (fun() -> 
         let k = key 
         key <- key + 1 
         subscriptions <- subscriptions.Add(k, o) 
         k) 
       { new IDisposable with 
        member this.Dispose() = 
         lock thisLock (fun() -> 
          subscriptions <- subscriptions.Remove(k)) } } 
    let mutable finished = false 
    // The methods below are not thread-safe; the source ought not call these methods concurrently 
    member this.Next(x) = 
     Debug.Assert(not finished, "IObserver is already finished") 
     next x 
    member this.Completed() = 
     Debug.Assert(not finished, "IObserver is already finished") 
     finished <- true 
     completed() 
    member this.Error(e) = 
     Debug.Assert(not finished, "IObserver is already finished") 
     finished <- true 
     error e 
    // The object returned here is threadsafe; you can subscribe and unsubscribe (Dispose) concurrently from multiple threads 
    member this.Value = obs 

Me interesarán las ideas sobre lo que está bien o mal aquí; No he tenido la oportunidad de mirar a todas las cosas nuevas Rx de devlabs sin embargo ...

Mis propias experiencias sugieren que:

• Aquellos que se suscriban a los observables nunca se debe tirar de las suscripciones. No hay nada razonable que un observable pueda hacer cuando lanza un suscriptor. (Esto es similar a los eventos.) Lo más probable es que la excepción simplemente suba a un controlador catch-all de nivel superior o bloquee la aplicación.
• Las fuentes probablemente deberían ser "lógicamente de un solo hilo". Creo que puede ser más difícil escribir clientes que puedan reaccionar a llamadas concurrentes OnNext; incluso si cada llamada individual proviene de un hilo diferente, es útil evitar llamadas concurrentes.
• Definitivamente es útil tener una clase base/ayudante que imponga algunos 'contratos'.

estoy muy ansioso por ver si la gente puede mostrar un asesoramiento más concreto a lo largo de estas líneas.

Answer 2

1. grieta abierta Reflector y echar un vistazo.

2. ver algunos videos C9 - this uno muestra cómo se puede 'deriva' Select 'Combinator'

3. El secreto es crear clases AnonymousObservable, AnonymousObserver y AnonymousDisposable, (que son sólo el trabajo arounds por el hecho que no puedes crear instancias de interfaces). No contienen implementación, ya que lo transfieres con acciones y funciones.

Por ejemplo:

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); 
    } 
} 

dejaré que usted se resuelve el resto ... Es un muy buen ejercicio de comprensión.

Hay un pequeño hilo que crece here con preguntas relacionadas.

Answer 3

simplemente una observación con respecto a esta aplicación:

después colecciones concurrentes están introduciendo en FW .net 4 es probable que sea mejor utilizar ConcurrentDictioary en lugar de un simple diccionario.

guarda bloqueos de manejo en la colección.

adi.

Answer 4

Sí, la palabra clave de rendimiento es encantadora; tal vez habrá algo similar para IObservable (OfT)? [Editar:. En la PDC '09 talk Eric Meijer, dice "sí, ver este espacio" a un rendimiento declarativo para la generación de los observables]

algo cerca (en vez de rodar su propio), echa un vistazo a the bottom de la wiki "(not yet) 101 Rx Samples", donde el equipo sugiere el uso de la clase Subject (T) como un "backend" para implementar un IObservable (OfT). Aquí está su ejemplo:

public class Order 
{    
    private DateTime? _paidDate; 

    private readonly Subject<Order> _paidSubj = new Subject<Order>(); 
    public IObservable<Order> Paid { get { return _paidSubj.AsObservable(); } } 

    public void MarkPaid(DateTime paidDate) 
    { 
     _paidDate = paidDate;     
     _paidSubj.OnNext(this); // Raise PAID event 
    } 
} 

private static void Main() 
{ 
    var order = new Order(); 
    order.Paid.Subscribe(_ => Console.WriteLine("Paid")); // Subscribe 

    order.MarkPaid(DateTime.Now); 
} 

Answer 5

El official documentation deprecates usuarios ejecución IObservable sí mismos. En lugar de ello, se espera que los usuarios utilicen el método de fábrica Observable.Create

Cuando sea posible, implementar nuevos operadores mediante la composición de los operadores existentes. De lo contrario, aplicar operadores personalizados utilizando Observable.Create

Sucede que Observable.Create es una envoltura alrededor de la clase trivial interna del reactivo AnonymousObservable:

public static IObservable<TSource> Create<TSource>(Func<IObserver<TSource>, IDisposable> subscribe) 
{ 
    if (subscribe == null) 
    { 
     throw new ArgumentNullException("subscribe"); 
    } 
    return new AnonymousObservable<TSource>(subscribe); 
} 

No sé por qué no hacen su aplicación público, pero bueno, lo que sea.