fix(sqs,sns): from with a list of objects

kestra-io · Aug 22, 2024 · b8f9a0b · b8f9a0b
1 parent 90e8fd3
commit b8f9a0b
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 4 deletions.
diff --git a/src/main/java/io/kestra/plugin/aws/sns/Publish.java b/src/main/java/io/kestra/plugin/aws/sns/Publish.java
@@ -83,8 +83,8 @@ public Publish.Output run(RunContext runContext) throws Exception {
 
             } else if (this.from instanceof List) {
                 flowable = Flux
-                    .fromArray(((List<Message>) this.from).toArray())
-                    .cast(Message.class);
+                    .fromIterable((List<?>) this.from)
+                    .map(map -> JacksonMapper.toMap(map, Message.class));
 
                 resultFlowable = this.buildFlowable(flowable, snsClient, topicArn, runContext);
 

diff --git a/src/main/java/io/kestra/plugin/aws/sqs/Publish.java b/src/main/java/io/kestra/plugin/aws/sqs/Publish.java
@@ -22,6 +22,8 @@
 import java.io.InputStreamReader;
 import java.net.URI;
 import java.util.List;
+import java.util.Map;
+
 import jakarta.validation.constraints.NotNull;
 
 import static io.kestra.core.utils.Rethrow.throwFunction;
@@ -85,8 +87,8 @@ public Output run(RunContext runContext) throws Exception {
 
             } else if (this.from instanceof List) {
                 flowable = Flux
-                    .fromArray(((List<Message>) this.from).toArray())
-                    .cast(Message.class);
+                    .fromIterable((List<?>) this.from)
+                    .map(map -> JacksonMapper.toMap(map, Message.class));
 
                 resultFlowable = this.buildFlowable(flowable, sqsClient, queueUrl, runContext);