Merge pull request #61 from kestra-io/fix/fix-npe-with-result-flowabl…

…e-block fix(): changed block() to blockOptional()
kestra-io · Sep 5, 2024 · 09ae9ca · 09ae9ca
2 parents c3ac68f + 0eb6230
commit 09ae9ca
Showing 1 changed file with 2 additions and 6 deletions.
diff --git a/src/main/java/io/kestra/plugin/amqp/Publish.java b/src/main/java/io/kestra/plugin/amqp/Publish.java
@@ -107,9 +107,7 @@ public Publish.Output run(RunContext runContext) throws Exception {
                     flowable = FileSerde.readAll(inputStream, Message.class);
                     resultFlowable = this.buildFlowable(flowable, channel, runContext);
 
-                    count = resultFlowable
-                        .reduce(Integer::sum)
-                        .block();
+                    count = resultFlowable.reduce(Integer::sum).blockOptional().orElse(0);
                 }
 
             } else if (this.from instanceof List) {
@@ -128,9 +126,7 @@ public Publish.Output run(RunContext runContext) throws Exception {
 
                 resultFlowable = this.buildFlowable(flowable, channel, runContext);
 
-                count = resultFlowable
-                    .reduce(Integer::sum)
-                    .block();
+                count = resultFlowable.reduce(Integer::sum).blockOptional().orElse(0);
             } else {
                 publish(channel, JacksonMapper.toMap(runContext.render((Map<String, Object>) this.from), Message.class), runContext);
             }