¿Cómo obtengo un IObservable para enviar el último valor al momento de la suscripción?

Question

Por lo general, cuando se suscribe a los cambios de un valor, también está interesado en conocer el valor inicial. Quiero que IObservable almacene en caché el valor más reciente (o inicial) e inserte ese valor en la suscripción.

Al utilizar eventos de civil que a menudo terminan con el código que se parece a

x.SomeEvent += SomeEventHandler; 
SomeEventHandler(x, EventArgs.Empty); 

Usando IObservable que estaba esperando para envolver el evento con algo que empuja el valor inicial. Si tengo varios suscriptores que deben recibir el valor más nuevo momento de la suscripción tengo un código que funciona bien si me suscribo a la derecha después de crear el IObservable pero no si se activa el evento antes de suscribirse:

class Program 
{ 
    static void Main() 
    { 
     var s = new Source { Value = 1 }; 
     var values = Observable.Return(s.Value).Concat(
      Observable.FromEvent(
       h => s.ValueChanged += h, 
       h => s.ValueChanged -= h) 
      .Select(_ => s.Value)); 
     using (values.Subscribe(Console.WriteLine)) 
     { 
      s.Value = 2; // prints 1,2 as expected 
     } 
     using (values.Subscribe(Console.WriteLine)) 
     { 
      s.Value = 3; // prints 1,3 - expected 2,3 
     } 
    } 
} 

class Source 
{ 
    private int _value; 
    public int Value 
    { 
     get { return _value; } 
     set 
     { 
      if (_value == value) 
       return; 
      _value = value; 
      if (ValueChanged != null) 
       ValueChanged(this, EventArgs.Empty); 
     } 
    } 

    public event EventHandler ValueChanged; 
} 

¿Cómo se crea una IObservable que funciona como se esperaba?

Answer 1

La solución es suscribir un BehaviorSubject al observable, y suscribir todos los observadores al BehaviorSubject. El BehaviorSubject recordará la última notificación y notificará a los nuevos observadores al momento de la suscripción.

Eche un vistazo al método de extensión Observable.Publish que tiene un parámetro initialValue. Esto crea un IConnectableObservable que internamente usa un BehaviorSubject.

var s = new Source { Value = 1 }; 

var values = Observable.FromEvent(h => s.ValueChanged += h, 
            h => s.ValueChanged -= h) 
         .Select(e => e.NewValue) 
         .Publish(s.Value); 

using (values.Connect())       // subscribes subject to event 
{ 
    using (values.Subscribe(Console.WriteLine)) // subscribes to subject 
    { 
     s.Value = 2; 
    }           // unsubscribes from subject 

    using (values.Subscribe(Console.WriteLine)) // subscribes to subject 
    { 
     s.Value = 3; 
    }           // unsubscribes from subject 

}           // unsubscribes subject from event 

(no probado)