feat: use FileSerde.writeAll and buffering for improved performances

src/main/java/io/kestra/plugin/redis/list/ListPop.java

@@ -13,6 +13,8 @@
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.io.*;
 import java.net.URI;
@@ -21,6 +23,8 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static io.kestra.core.utils.Rethrow.throwFunction;
+
 @SuperBuilder
 @ToString
 @EqualsAndHashCode
@@ -71,18 +75,17 @@ public Output run(RunContext runContext) throws Exception {
                 throw new Exception("maxDuration or maxRecords must be set to avoid infinite loop");
             }
 
-            try (BufferedOutputStream output = new BufferedOutputStream(new FileOutputStream(tempFile))) {
+            try (var output = new BufferedWriter(new FileWriter(tempFile), FileSerde.BUFFER_SIZE)) {
                 AtomicInteger total = new AtomicInteger();
                 ZonedDateTime started = ZonedDateTime.now();
 
                 boolean empty;
                 do {
                     List<String> data = factory.listPop(key, count);
                     empty = data.isEmpty();
-                    for (String str : data) {
-                        FileSerde.write(output, this.serdeType.deserialize(str));
-                        total.getAndIncrement();
-                    }
+                    var flux = Flux.fromIterable(data).map(throwFunction(str -> this.serdeType.deserialize(str)));
+                    Mono<Long> longMono = FileSerde.writeAll(output, flux);
+                    total.addAndGet(longMono.block().intValue());
                 }
                 while (!this.ended(empty, total, started));