¿Las extensiones reactivas son compatibles con los búfers de laminación?

Question

estoy usando extensiones de reactivos para cotejar los datos en memorias intermedias de 100 ms:

this.subscription = this.dataService 
    .Where(x => !string.Equals("FOO", x.Key.Source)) 
    .Buffer(TimeSpan.FromMilliseconds(100)) 
    .ObserveOn(this.dispatcherService) 
    .Where(x => x.Count != 0) 
    .Subscribe(this.OnBufferReceived); 

Esto funciona bien. Sin embargo, quiero un comportamiento ligeramente diferente al proporcionado por la operación Buffer. Esencialmente, quiero restablecer el temporizador si se recibe otro elemento de datos. Solo cuando no se hayan recibido datos para los 100 ms completos, quiero manejarlos. Esto abre la posibilidad de nunca manejar los datos, por lo que también debería poder especificar un recuento máximo. Me imagino algo en la línea de:

.SlidingBuffer(TimeSpan.FromMilliseconds(100), 10000) 

que he tenido una mirada alrededor y no he podido encontrar nada parecido en Rx? ¿Alguien puede confirmar/negar esto?

Answer 1

Escribí una extensión para hacer la mayor parte de lo que está buscando - BufferWithInactivity.

Aquí está:

public static IObservable<IEnumerable<T>> BufferWithInactivity<T>(
    this IObservable<T> source, 
    TimeSpan inactivity, 
    int maximumBufferSize) 
{ 
    return Observable.Create<IEnumerable<T>>(o => 
    { 
     var gate = new object(); 
     var buffer = new List<T>(); 
     var mutable = new SerialDisposable(); 
     var subscription = (IDisposable)null; 
     var scheduler = Scheduler.ThreadPool; 

     Action dump =() => 
     { 
      var bts = buffer.ToArray(); 
      buffer = new List<T>(); 
      if (o != null) 
      { 
       o.OnNext(bts); 
      } 
     }; 

     Action dispose =() => 
     { 
      if (subscription != null) 
      { 
       subscription.Dispose(); 
      } 
      mutable.Dispose(); 
     }; 

     Action<Action<IObserver<IEnumerable<T>>>> onErrorOrCompleted = 
      onAction => 
      { 
       lock (gate) 
       { 
        dispose(); 
        dump(); 
        if (o != null) 
        { 
         onAction(o); 
        } 
       } 
      }; 

     Action<Exception> onError = ex => 
      onErrorOrCompleted(x => x.OnError(ex)); 

     Action onCompleted =() => onErrorOrCompleted(x => x.OnCompleted()); 

     Action<T> onNext = t => 
     { 
      lock (gate) 
      { 
       buffer.Add(t); 
       if (buffer.Count == maximumBufferSize) 
       { 
        dump(); 
        mutable.Disposable = Disposable.Empty; 
       } 
       else 
       { 
        mutable.Disposable = scheduler.Schedule(inactivity,() => 
        { 
         lock (gate) 
         { 
          dump(); 
         } 
        }); 
       } 
      } 
     }; 

     subscription = 
      source 
       .ObserveOn(scheduler) 
       .Subscribe(onNext, onError, onCompleted); 

     return() => 
     { 
      lock (gate) 
      { 
       o = null; 
       dispose(); 
      } 
     }; 
    }); 
} 

Answer 2

creo que esto puede ser implementado en la parte superior del Método de regulación como se muestra a continuación:

public static IObservable<IList<T>> SlidingBuffer<T>(this IObservable<T> obs, TimeSpan span, int max) 
     { 
      return Observable.CreateWithDisposable<IList<T>>(cl => 
      { 
       var acc = new List<T>(); 
       return obs.Buffer(span) 
         .Subscribe(next => 
         { 
          if (next.Count == 0) //no activity in time span 
          { 
           cl.OnNext(acc); 
           acc.Clear(); 
          } 
          else 
          { 
           acc.AddRange(next); 
           if (acc.Count >= max) //max items collected 
           { 
            cl.OnNext(acc); 
            acc.Clear(); 
           } 
          } 
         }, err => cl.OnError(err),() => { cl.OnNext(acc); cl.OnCompleted(); }); 
      }); 
     } 

NOTA: Yo no lo he probado, pero espero que sea te da la idea.

Answer 3

Esto es posible combinando los métodos integrados Window y Throttle de Observable. En primer lugar, vamos a resolver el problema más sencillo cuando ignoramos la condición de máxima recuento:

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay) 
{ 
    var closes = stream.Throttle(delay); 
    return stream.Window(() => closes).SelectMany(window => window.ToList()); 
} 

El potente Window method hicieron el trabajo pesado. Ahora es bastante fácil ver cómo agregar un recuento máximo:

public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? max=null) 
{ 
    var closes = stream.Throttle(delay); 
    if (max != null) 
    { 
     var overflows = stream.Where((x,index) => index+1>=max); 
     closes = closes.Merge(overflows); 
    } 
    return stream.Window(() => closes).SelectMany(window => window.ToList()); 
} 

Voy a escribir una publicación explicando esto en mi blog. https://gist.github.com/2244036

documentación para el método de la ventana:

• http://leecampbell.blogspot.co.uk/2011/03/rx-part-9join-window-buffer-and-group.html
• http://enumeratethis.com/2011/07/26/financial-charts-reactive-extensions/

Answer 4

con extensiones 2.0 Rx, su puede responder a ambas necesidades con un nuevo buffer sobrecargue la aceptación de un tiempo de espera y una talla:

this.subscription = this.dataService 
    .Where(x => !string.Equals("FOO", x.Key.Source)) 
    .Buffer(TimeSpan.FromMilliseconds(100), 1) 
    .ObserveOn(this.dispatcherService) 
    .Where(x => x.Count != 0) 
    .Subscribe(this.OnBufferReceived); 

Consulte https://msdn.microsoft.com/en-us/library/hh229200(v=vs.103).aspx para la documentación.