RX: ráfagas de elementos de grupo / lote en una secuencia observable
Tengo una secuencia observable. Cuando se inserta el primer elemento,Me gustaría iniciar un temporizador y agrupar elementos insertados posteriormente durante el intervalo de tiempo del temporizador. Entonces, el temporizador no volvería a comenzar hasta que se inserte otro elemento en la secuencia.
Entonces algo como esto:
--------|=====timespan====|---------------|=====timespan====|-------------->
1 2 3 4 5 6 7 8
produciría:
[1,2,3,4,5], [6,7,8]
Intenté con Observable.Buffer () y un intervalo de tiempo, pero desde mi experimentación, puedo ver que el temporizador se inicia tan pronto como nos suscribimos a la secuencia observable y se reinicia tan pronto como se completa el temporizador anterior.
Entonces, teniendo la misma secuencia que el ejemplo anterior y usando el Buffer () con un intervalo de tiempo, tendría algo como esto:
|=====timespan====|=====timespan====|=====timespan====|=====timespan====|-->
1 2 3 4 5 6 7 8
que produciría esto:
[1,2,3,4], [5], [6,7], [8]
Así es como probé este comportamiento con el Buffer:
var source = Observable.Concat(Observable.Timer(TimeSpan.FromSeconds(6)).Select(o => 1),
Observable.Timer(TimeSpan.FromSeconds(1)).Select(o => 2),
Observable.Timer(TimeSpan.FromSeconds(3)).Select(o => 3),
Observable.Never<int>());
Console.WriteLine("{0} => Started", DateTime.Now);
source.Buffer(TimeSpan.FromSeconds(4))
.Subscribe(i => Console.WriteLine("{0} => [{1}]", DateTime.Now, string.Join(",", i)));
Con la salida:
4/24/2015 7:01:09 PM => Started
4/24/2015 7:01:13 PM => []
4/24/2015 7:01:17 PM => [1,2]
4/24/2015 7:01:21 PM => [3]
4/24/2015 7:01:25 PM => []
4/24/2015 7:01:29 PM => []
4/24/2015 7:01:33 PM => []
¿Alguien tiene una idea sobre cómo hacer esto? ¡Gracias por adelantado!