fix(): changed block() to blockOptional()

kestra-io · Sep 5, 2024 · 0eb6230 · 0eb6230
1 parent c3ac68f
commit 0eb6230
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);
             }