Reaktor 3 'Intervallpuffer' Flux?

Question

Wie kann ich vorhandene Flux-Operatoren verwenden, damit ein Flux eingehende Werte in mehrere Listen mit einer minimalen Verzögerung zwischen den Retouren zurückgibt?

Answer 1

Ich dachte buffer(Duration) Ihren Bedarf passen würde, aber es funktioniert nicht.

bearbeiten: Lassen Sie dies für den Fall, dass jemand mit Ihrem genau gleichen Bedarf versucht ist, diesen Operator zu verwenden. Diese Variante des Puffers teilt die Sequenz in aufeinanderfolgende Zeitfenster auf (die jeweils eine buffer erzeugen). Das heißt, die neue delay beginnt am Ende der vorherigen, nicht wenn ein neues Element ohne Verzögerung ausgegeben wird.

Answer 2

Dies kann mit einem nicht trivialen Satz von zusammengesetzten Operatoren erreicht werden.

import java.time.Duration; 
import java.util.*; 

import reactor.core.publisher.*; 

public class DelayedBuffer { 

    public static void main(String[] args) { 
     Flux.just(1, 2, 3, 6, 7, 10) 
     .flatMap(v -> Mono.delayMillis(v * 1000) 
       .doOnNext(w -> System.out.println("T=" + v)) 
       .map(w -> v) 
     ) 
     .compose(f -> delayedBufferAfterFirst(f, Duration.ofSeconds(2))) 
     .doOnNext(System.out::println) 
     .blockLast(); 
    } 

    public static <T> Flux<List<T>> delayedBufferAfterFirst(Flux<T> source, Duration d) { 
     return source 
     .publish(f -> { 
      return f.take(1).collectList() 
      .concatWith(f.buffer(d).take(1)) 
      .repeatWhen(r -> r.takeUntilOther(f.ignoreElements())); 
     }); 
    } 
} 

(Beachten Sie jedoch, dass die zu erwartenden Emissionsmuster besser mit einem benutzerdefinierten Bediener angepasst werden können aufgrund von Zeit beteiligt zu sein.)