リアクタ3 'インターバルバッファ' Flux？

Question

既存のFlux演算子を使用して、Fluxが入力値を複数のリストに返すようにするにはどうすればよいでしょうか？

Answer 1

私はbuffer(Duration)があなたのニーズに合うだろうと思ったが、それはしていません。

編集：まったく同じ必要性を持つ人が、そのオペレータを使用するように誘惑されている場合は、このままにしてください。バッファのこの変形は、連続する時間ウィンドウ（それぞれがbufferを生成する）にシーケンスを分割する。つまり、新しいdelayは、前のものの最後から開始します。新しい遅延のない要素が放出されるとは限りません。

Answer 2

これは、単純ではない組の演算子で実現できます。

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())); 
     }); 
    } 
} 

（予想される放射パターンが良好期限が関与しているにカスタムオペレータと一致させることができることは留意されたい。）