Estoy buscando un selector observable con una firma similar a esta:
static IObservable<T> TakeLatest(this IObservable<T> input, TimeSpan interval)Que debería:
En términos de canicas, algo como lo siguiente, asumiendo intervalo = 2 unidades de tiempo:
| Tiempo | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| Aporte | UN | B | C | D | mi | F (completa) | |||||||||
| Producción | UN | B | D | D | mi | mi | completo (F ya no se emite) |
¿Hay alguna forma lista para usar de hacerlo, o un selector razonablemente fácil para producir estos resultados?
Esto probablemente debería hacer exactamente lo que quieres. Aunque no lo he probado.
/// <summary>Samples the source observable sequence at each interval, /// allowing repeated emissions of the same element.</summary> public static IObservable<T> SampleWithDuplicates<T>(this IObservable<T> source, TimeSpan interval, IScheduler scheduler = null) { scheduler ??= DefaultScheduler.Instance; return source.Publish(published => Observable .Interval(interval, scheduler) .WithLatestFrom(published, (_, x) => x) .Merge(published.FirstAsync()) .TakeUntil(published.LastOrDefaultAsync())); }He hecho lo siguiente ahora: creo que funciona, pero lo dejaré abierto en caso de que alguien pueda pensar en una forma más elegante (o puede pensar en un problema con mi implementación actual)
static IObservable<T> TakeLatest(this IObservable<T> input, TimeSpan interval, IScheduler scheduler) => input .FirstAsync() .Select(_ => Observable.Interval(interval, scheduler).StartWith(0)) .Switch() .CombineLatest(input, (a,b) => (a,b)) .DistinctUntilChanged(x => xa) .Select(x => xb) .TakeUntil(input.LastAsync());