feat(amqp): allow an incomplete URL as parameter

src/main/java/io/kestra/plugin/amqp/AbstractAmqpConnection.java

@@ -4,6 +4,7 @@
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
+import java.util.Optional;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -32,11 +33,11 @@ public ConnectionFactory connectionFactory(RunContext runContext) throws Excepti
         }
 
         ConnectionFactory factory = new ConnectionFactory();
-        factory.setHost(runContext.render(host));
-        factory.setPort(Integer.parseInt(runContext.render(port)));
-        factory.setUsername(runContext.render(username));
-        factory.setPassword(runContext.render(password));
-        factory.setVirtualHost(runContext.render(virtualHost));
+        Optional.ofNullable(runContext.render(host)).ifPresent(factory::setHost);
+        Optional.ofNullable(runContext.render(port)).map(Integer::parseInt).ifPresent(factory::setPort);
+        Optional.ofNullable(runContext.render(username)).ifPresent(factory::setUsername);
+        Optional.ofNullable(runContext.render(password)).ifPresent(factory::setPassword);
+        Optional.ofNullable(runContext.render(virtualHost)).ifPresent(factory::setVirtualHost);
 
         factory.setExceptionHandler(new AmqpExceptionHandler(runContext.logger()));
 
@@ -47,14 +48,18 @@ void parseFromUrl(RunContext runContext, String url) throws IllegalVariableEvalu
         URI amqpUri = new URI(runContext.render(url));
 
         host = amqpUri.getHost();
-        port = String.valueOf(amqpUri.getPort());
+        if (amqpUri.getPort() != -1) {
+            port = String.valueOf(amqpUri.getPort());
+        }
 
         String auth = amqpUri.getUserInfo();
-        int pos = auth.indexOf(':');
-        username = pos > 0 ? auth.substring(0, pos) : auth;
-        password = pos > 0 ? auth.substring(pos + 1) : "";
+        if (auth != null) {
+            int pos = auth.indexOf(':');
+            username = pos > 0 ? auth.substring(0, pos) : auth;
+            password = pos > 0 ? auth.substring(pos + 1) : "";
+        }
 
-        if (!amqpUri.getPath().equals("")) {
+        if (!amqpUri.getPath().isEmpty()) {
             virtualHost = amqpUri.getPath();
         }
     }

src/test/java/io/kestra/plugin/amqp/AMQPTest.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.core.JsonParseException;
 import com.google.common.collect.ImmutableMap;
+import com.rabbitmq.client.ConnectionFactory;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.serializers.FileSerde;
@@ -21,8 +22,6 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
@@ -37,6 +36,69 @@ class AMQPTest {
     @Inject
     protected StorageInterface storageInterface;
 
+    @Test
+    void createConnectionFactoryWithDefaultValues() throws Exception {
+        Publish push = Publish.builder().build();
+
+        ConnectionFactory connectionFactory = push.connectionFactory(runContextFactory.of());
+        assertThat(connectionFactory.getHost(), is(ConnectionFactory.DEFAULT_HOST));
+        assertThat(connectionFactory.getPort(), is(ConnectionFactory.DEFAULT_AMQP_PORT));
+        assertThat(connectionFactory.getUsername(), is(ConnectionFactory.DEFAULT_USER));
+        assertThat(connectionFactory.getPassword(), is(ConnectionFactory.DEFAULT_PASS));
+        assertThat(connectionFactory.getVirtualHost(), is(ConnectionFactory.DEFAULT_VHOST));
+    }
+
+    @Test
+    void createConnectionFactoryWithUriOnly() throws Exception {
+        Publish push = Publish.builder()
+                .url("amqp://kestra:[email protected]:12345/my_vhost")
+                .build();
+
+        ConnectionFactory connectionFactory = push.connectionFactory(runContextFactory.of());
+        assertThat(connectionFactory.getHost(), is("example.org"));
+        assertThat(connectionFactory.getPort(), is(12345));
+        assertThat(connectionFactory.getUsername(), is("kestra"));
+        assertThat(connectionFactory.getPassword(), is("K3str4"));
+        assertThat(connectionFactory.getVirtualHost(), is("/my_vhost"));
+    }
+
+    @Test
+    void createConnectionFactoryWithFieldsOnly() throws Exception {
+        Publish push = Publish.builder()
+                .host("example.org")
+                .port("12345")
+                .username("kestra")
+                .password("K3str4")
+                .virtualHost("/my_vhost")
+                .build();
+
+        ConnectionFactory connectionFactory = push.connectionFactory(runContextFactory.of());
+        assertThat(connectionFactory.getHost(), is("example.org"));
+        assertThat(connectionFactory.getPort(), is(12345));
+        assertThat(connectionFactory.getUsername(), is("kestra"));
+        assertThat(connectionFactory.getPassword(), is("K3str4"));
+        assertThat(connectionFactory.getVirtualHost(), is("/my_vhost"));
+    }
+
+    @Test
+    void createConnectionFactoryWithUriAndFields() throws Exception {
+        Publish push = Publish.builder()
+                .url("amqp://example.org") // values extracted from the URI are considered first
+                .host("ignore.it")
+                .port("12345")
+                .username("kestra")
+                .password("K3str4")
+                .virtualHost("/my_vhost")
+                .build();
+
+        ConnectionFactory connectionFactory = push.connectionFactory(runContextFactory.of());
+        assertThat(connectionFactory.getHost(), is("example.org"));
+        assertThat(connectionFactory.getPort(), is(12345));
+        assertThat(connectionFactory.getUsername(), is("kestra"));
+        assertThat(connectionFactory.getPassword(), is("K3str4"));
+        assertThat(connectionFactory.getVirtualHost(), is("/my_vhost"));
+    }
+
     @Test
     void pushAsList() throws Exception {
         Publish push = Publish.builder()