search-flux-buffer-parallel.java

import reactor.core.publisher.*;
import java.nio.file.Path;

public Flux<String> searchReactiveParallel(Integer buffer, Integer threads) {

        Path ipPath = Path.of(pathFile);

        if (buffer == null)
            return Flux.using(
                    () -> Files.lines(ipPath),
                    Flux::fromStream,
                    Stream::close)
                    .subscribeOn(Schedulers.newParallel("searchReactive2", threads))
                    .share()
                    .filter(l -> l.contains(SEARCH_TOKEN))
                    .map(s -> s + "\r\n");

        else
            return Flux.using(
                    () -> Files.lines(ipPath),
                    Flux::fromStream,
                    Stream::close)
                    .subscribeOn(Schedulers.newParallel("searchReactive2", threads))
                    .share()
                    .buffer(buffer)
                    .flatMap(l -> Flux.fromStream(
                            l.stream().filter(line -> line.contains(SEARCH_TOKEN)).map(s -> s + "\r\n")
                            )
                    );
    }

related to medium article https://gae-piaz.medium.com/search-string-in-file-with-java-as-fast-as-possible-fedafdc7a3ee