Rx Marco: ejecutar una acción en tiempo de espera sin interrumpir la secuencia original, observable

Question

Dada una fuente observable, generada mediante el sondeo de los cambios (a) estado de un dispositivo de bajo nivel ...

// observable source metacode: 
IObservable<DeviceState> source = Observable.Interval(TimeSpan.FromSeconds(0.5)) 
    .Select(tick => new DeviceState(_device.ReadValue())) 
    .DistinctUntilChanged(); 

.. . y un consumidor que actualiza la interfaz de usuario ...

// UI metacode: 
service.GetObservableDeviceStates() 
    .Subscribe(state => viewModel.CurrentState = state.ToString()); 

... necesito para ejecutar una acción personalizada después de x segundos de "inactividad" de la fuente, sin interrumpir la suscripción a la fuente. Algo como esto:

// UI metacode: 
service.GetObservableDeviceStates() 
    .DoOnTimeout(TimeSpan.FromSeconds(x),() => viewModel.CurrentState = "Idle") 
    .Subscribe(state => viewModel.CurrentState = state.ToString()); 

¿Cuáles son las mejores prácticas? Las posibles soluciones que vienen a la mente son (soy un novato Rx):

1. Buffer (aunque no es tan legible)
2. Jugando this Timeout overload;

3. Volviendo algo especial "del lado del servicio" cuando nada cambia (en lugar de utilizar DistinctUntilChanged) y tratar con ella en el código de interfaz de usuario:

   service.GetObservableDeviceStates() .Subscribe (estado => viewModel.CurrentState = state.Special? "Idle": state.ToString());

EDIT: como se informó in the answer, la solución es:

 service.GetObservableDeviceStates() 
      .Do(onNext) 
      .Throttle(TimeSpan.FromSeconds(x)) 
      .Subscribe(onTimeout); 

Edit2 (advertencia)

Si los componentes actualizaciones de interfaz de usuario onNext y onTimeout, para evitar CrossThreadExceptions dos ObserveOn (uiSynchronizationContext) son necesarios, ya que Throttle trabaja en otro hilo.

 service.GetObservableDeviceStates() 
      .ObserveOn(uiSynchronizationContext) 
      .Do(onNext) 
      .Throttle(TimeSpan.FromSeconds(x)) 
      .ObserveOn(uiSynchronizationContext) 
      .Subscribe(onTimeout); 

Answer 1

tiempo de espera es más o menos significaba para observaciones que representan operaciones asíncronas individuales - por ejemplo, para devolver un valor predeterminado o si dicho OnError observables no le ha notificado en una cierta cantidad de tiempo.

El operador que está buscando es Throttle, aunque al principio no lo parezca. Throttle(p) le proporciona una secuencia que produce un valor cuando la secuencia fuente no ha producido un valor para el período p.

Paralelo a su código existente, puede usar source.Throttle(period).Do(...side effect).

Answer 2

Personalmente evitaría el método Do para esto. Hace que el código en este ejemplo sea bastante fácil, pero encuentro que una vez que el uso de 'Hacer' se cuela en la base de código, pronto tiene spaghetti.

También podría considerar el uso de combinaciones de Amb, Timer, TakeUntil, Throttle, etc. para obtener el resultado que busca y seguir manteniendo el Monad *. O, en términos simples, supongo que idealmente quieres tener una secuencia de los valores de estado que llegan y no requerir la necesidad de poner un temporizador en tu código (es decir, descargarlo al servicio).

public IObservable<DeviceStatus> GetObservableDeviceStates(TimeSpan silencePeriod) 
{ 
    return Observable.Create<DeviceStatus>(
    o=> 
    { 
     var idle = Observable.Timer(silencePeriod).Select(_=>new DeviceStatus("Idle")); 

     var polledStatus = Observable.Interval(TimeSpan.FromSeconds(0.5)) 
         .Select(tick => new DeviceStatus(_device.ReadValue())) 
         .DistinctUntilChanged() 
         .Publish(); 

     var subscription = (from status in polledStatus 
          from cont in Observable.Return(status).Concat(idle.TakeUntil(polledStatus)) 
          select cont) 
        .Subscribe(o); 

     return new CompositeDisposable(subscription, polledStatus.Connect()); 
    }); 
} 

Este código tiene ahora el Servicio de devolver un valor de estado de reposo una vez se ha producido el período especificado de cambio de silencio.

Esto significa que el código de interfaz de usuario meta se mantiene simple y la lógica relacionada con devicestatus permanezca donde pertenece

// UI metacode: 
service.GetObservableDeviceStates(TimeSpan.FromSeconds(2)) 
    .Subscribe(state => viewModel.CurrentState = state.ToString());