BufferWithTime de Reactive Extensions va-t-il invoquer OnNext avec chevauchement?

Question

J'utilise BufferWithTime() de Rx pour traiter en lot des messages. Si ma méthode OnNext prend plus de temps à terminer que l'intervalle de temps utilisé, puis-je me retrouver avec deux invocations simultanées de mes méthodes OnNext? En d'autres termes, l'intervalle de temps spécifié dans l'appel BufferWithTime() sera-t-il évalué en termes absolus ou agira-t-il comme un «sommeil asynchrone» après le retour de ma méthode OnNext?

Answer 1

Non, les appels à OnNext seront placés en file d'attente. Le code écrit ci-dessous pour la trace lorsque le temps de maintien est écoulée, ainsi que le début/la fin de l'appel OnNext à l'abonné (dans laquelle il peut accueillir 4 fois le temps tampon):

Observable.Interval(TimeSpan.FromMilliseconds(50), Scheduler.TaskPool) 
    .Take(50) 
    .BufferWithTime(TimeSpan.FromMilliseconds(100)) 
    .Do(_ => Trace.WriteLine("Buffer elapsed")) 
    .ObserveOn(Scheduler.TaskPool) 
    .Subscribe(_ => 
    { 
     Trace.WriteLine("Begin OnNext"); 
     Thread.Sleep(200); 
     Trace.WriteLine("End OnNext"); 
    }); 

La sortie est comme suit. Vous pouvez voir que Begin/End OnNext chevauche jamais, même avec Buffer elapsed se produit deux fois au cours d'un appel OnNext:

tampon écoula
Begin OnNext
tampon écoula
Tampon écoula
Fin OnNext
Begin OnNext
Tampon écoulé
Tampon écoulé
Tampon écoulé
Fin OnNext
Begin OnNext
Fin OnNext
Begin OnNext
Fin OnNext
Begin OnNext
Fin OnNext
Begin OnNext
Fin OnNext