Estensioni reattive: acceleratore/campione con intervallo variabile

Question

Ho uno IObservable che produce valori a intervalli casuali e voglio limitare questa sequenza. Una cosa che ho scoperto è che la definizione di "throttling" dell'operatore Throttle non è la mia.

Throttle produce valori solo dopo l'intervallo specificato con il silenzio (produce l'ultimo valore visualizzato). Pensavo che il throttling avrebbe significato produrre valori all'intervallo specificato (a meno che non ci sia il silenzio, ovviamente).

Dire, mi aspettavo che Observable.Interval(100).Select((_,i) => i).Throttle(200) produca (modulo qualsiasi problema di prestazioni/temporizzazione) i numeri pari, dal momento che lo sto riducendo a "mezza velocità". Tuttavia tale sequenza non produce alcun valore, perché non c'è mai un periodo di silenzio di lunghezza 200.

Così, ho scoperto che lo Sample effettivamente fa il comportamento "throttling" che voglio. Observable.Interval(100).Select((_,i) => i).Sample(200) produce (ancora, modulo qualsiasi problema di prestazioni/temporizzazione) la sequenza di numeri pari.

Tuttavia, ho un altro problema: l'intervallo varia a seconda dell'ultimo valore "campionato". Quello che voglio è quello di scrivere un operatore che assomiglia a questo:

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector); 

Il parametro intervalSelector produce l'intervallo per il campione successivo, e il primo campione ... o è presa al primo valore o da un parametro aggiuntivo , Non mi interessa

Ho provato a scrivere questo, ma mi sono ritrovato con una grande costruzione contorta che non funzionava correttamente. La mia domanda è, posso costruire questo utilizzando gli operatori esistenti (aka, con una sola linea)?

Answer 1

Molte ore dopo, e con un po 'di sonno, ho capito.

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector) 
{ 
    return source.TimeInterval() 
       .Scan(Tuple.Create(TimeSpan.Zero, false, default(T)), (acc, v) => 
       { 
        if(v.Interval >= acc.Item1) 
        { 
         return Tuple.Create(intervalSelector(v.Value), true, v.Value); 
        } 
        return Tuple.Create(acc.Item1 - v.Interval, false, v.Value); 
       }) 
       .Where(t => t.Item2) 
       .Select(x => x.Item3); 
} 

questo funziona come voglio: ogni volta che si produce un valore x, si smette di produrre valori fino intervalSelector(x) il tempo passa.

Answer 2

Non è quello che stai cercando for Observable.BufferWithTime?