Create SQS sender for imperative projects

README.md

@@ -318,7 +318,7 @@ The **`generateDrivenAdapter | gda`** task will generate a module in Infrastruct
    | mq                                  | JMS MQ Client to send messages      |                                                    | ☑ | ☑ |
    | ktor                                | HTTP client for kotlin              |                                                    | ☐ | ☑ |
    | dynamodb                            | Dynamo DB adapter                   |                                                    | ☑ | ☑ |
-   | sqs                                 | SQS message sender                  |                                                    | ☑ |         |
+   | sqs                                 | SQS message sender                  |                                                    | ☑ | ☑ |
 
 
    _**This task will generate something like that:**_

src/main/java/co/com/bancolombia/factory/adapters/DrivenAdapterSQS.java

@@ -7,21 +7,24 @@
 import co.com.bancolombia.factory.ModuleBuilder;
 import co.com.bancolombia.factory.ModuleFactory;
 import co.com.bancolombia.factory.commons.GenericModule;
-import co.com.bancolombia.factory.validations.ReactiveTypeValidation;
 import java.io.IOException;
 
 public class DrivenAdapterSQS implements ModuleFactory {
 
   @Override
   public void buildModule(ModuleBuilder builder) throws IOException, CleanException {
-    builder.runValidations(ReactiveTypeValidation.class);
     GenericModule.addAwsBom(builder);
-    builder.setupFromTemplate("driven-adapter/sqs");
+    String template = "driven-adapter/sqs";
+    if (builder.isReactive()) {
+      template = "driven-adapter/sqs-reactive";
+    }
+    builder.setupFromTemplate(template);
     builder.appendToSettings("sqs-sender", "infrastructure/driven-adapters");
     builder
         .appendToProperties("adapter.sqs")
         .put("region", "us-east-1")
-        .put("queueUrl", "http://localhost:4566/000000000000/sample");
+        .put("queueUrl", "http://localhost:4566/000000000000/sample")
+        .put("endpoint", "http://localhost:4566 # For localstack only");
     String dependency = buildImplementationFromProject(builder.isKotlin(), ":sqs-sender");
     builder.appendDependencyToModule(APP_SERVICE, dependency);
   }

src/main/resources/driven-adapter/sqs-reactive/build.gradle.mustache

@@ -0,0 +1,6 @@
+dependencies {
+    implementation project(':model')
+    implementation 'org.springframework.boot:spring-boot-autoconfigure'
+    implementation 'org.apache.logging.log4j:log4j-api'
+    implementation 'software.amazon.awssdk:sqs'
+}

src/main/resources/driven-adapter/sqs-reactive/definition.json

@@ -0,0 +1,13 @@
+{
+  "folders": [
+    "infrastructure/driven-adapters/sqs-sender/src/test/{{language}}/{{packagePath}}/sqs/sender/config"
+  ],
+  "files": {},
+  "java": {
+    "driven-adapter/sqs-reactive/build.gradle.mustache": "infrastructure/driven-adapters/sqs-sender/build.gradle",
+    "driven-adapter/sqs-reactive/sqs-sender.java.mustache": "infrastructure/driven-adapters/sqs-sender/src/main/{{language}}/{{packagePath}}/sqs/sender/SQSSender.java",
+    "driven-adapter/sqs-reactive/sqs-sender-config.java.mustache": "infrastructure/driven-adapters/sqs-sender/src/main/{{language}}/{{packagePath}}/sqs/sender/config/SQSSenderConfig.java",
+    "driven-adapter/sqs-reactive/sqs-sender-properties.java.mustache": "infrastructure/driven-adapters/sqs-sender/src/main/{{language}}/{{packagePath}}/sqs/sender/config/SQSSenderProperties.java"
+  },
+  "kotlin": {}
+}

src/main/resources/driven-adapter/sqs-reactive/sqs-sender-config.java.mustache

@@ -0,0 +1,64 @@
+package {{package}}.sqs.sender.config;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProviderChain;
+import software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.SystemPropertyCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider;
+{{#metrics}}
+import software.amazon.awssdk.metrics.MetricPublisher;
+{{/metrics}}
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+
+import java.net.URI;
+
+@Configuration
+@ConditionalOnMissingBean(SqsAsyncClient.class)
+public class SQSSenderConfig {
+
+    {{#metrics}}
+    @Bean
+    public SqsAsyncClient configSqs(SQSSenderProperties properties, MetricPublisher publisher) {
+        return SqsAsyncClient.builder()
+                .endpointOverride(resolveEndpoint(properties))
+                .region(Region.of(properties.getRegion()))
+                .overrideConfiguration(o -> o.addMetricPublisher(publisher))
+                .credentialsProvider(getProviderChain())
+                .build();
+    }
+    {{/metrics}}
+    {{^metrics}}
+    @Bean
+    public SqsAsyncClient configSqs(SQSSenderProperties properties) {
+        return SqsAsyncClient.builder()
+                .endpointOverride(resolveEndpoint(properties))
+                .region(Region.of(properties.getRegion()))
+                .credentialsProvider(getProviderChain())
+                .build();
+    }
+    {{/metrics}}
+
+    private AwsCredentialsProviderChain getProviderChain() {
+        return AwsCredentialsProviderChain.builder()
+                .addCredentialsProvider(EnvironmentVariableCredentialsProvider.create())
+                .addCredentialsProvider(SystemPropertyCredentialsProvider.create())
+                .addCredentialsProvider(WebIdentityTokenFileCredentialsProvider.create())
+                .addCredentialsProvider(ProfileCredentialsProvider.create())
+                .addCredentialsProvider(ContainerCredentialsProvider.builder().build())
+                .addCredentialsProvider(InstanceProfileCredentialsProvider.create())
+                .build();
+    }
+
+    private URI resolveEndpoint(SQSSenderProperties properties) {
+        if (properties.getEndpoint() != null) {
+            return URI.create(properties.getEndpoint());
+        }
+        return null;
+    }
+}

src/main/resources/driven-adapter/sqs-reactive/sqs-sender-properties.java.mustache

@@ -0,0 +1,47 @@
+package {{package}}.sqs.sender.config;
+
+{{#lombok}}
+import lombok.Data;
+{{/lombok}}
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+{{#lombok}}
+@Data
+{{/lombok}}
+@Configuration
+@ConfigurationProperties(prefix = "adapter.sqs")
+public class SQSSenderProperties {
+    private String region;
+    private String queueUrl;
+    private String endpoint;
+{{^lombok}}
+
+    public SQSSenderProperties() {
+    }
+
+    public String getRegion() {
+        return region;
+    }
+
+    public void setRegion(String region) {
+        this.region = region;
+    }
+
+    public String getQueueUrl() {
+        return queueUrl;
+    }
+
+    public void setQueueUrl(String queueUrl) {
+        this.queueUrl = queueUrl;
+    }
+
+    public String getEndpoint() {
+        return endpoint;
+    }
+
+    public void setEndpoint(String endpoint) {
+        this.endpoint = endpoint;
+    }
+{{/lombok}}
+}

src/main/resources/driven-adapter/sqs-reactive/sqs-sender.java.mustache

@@ -0,0 +1,46 @@
+package {{package}}.sqs.sender;
+
+import {{package}}.sqs.sender.config.SQSSenderProperties;
+{{#lombok}}
+import lombok.AllArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+{{/lombok}}
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
+import software.amazon.awssdk.services.sqs.model.SendMessageResponse;
+
+@Service
+{{#lombok}}
+@Log4j2
+@AllArgsConstructor
+{{/lombok}}
+public class SQSSender /*implements SomeGateway*/ {
+{{^lombok}}
+    private static final org.apache.logging.log4j.Logger log = org.apache.logging.log4j.LogManager.getLogger(SQSSender.class);
+{{/lombok}}
+    private final SQSSenderProperties properties;
+    private final SqsAsyncClient client;
+{{^lombok}}
+
+    public SQSSender(SqsAsyncClient client, SQSSenderProperties properties) {
+        this.client = client;
+        this.properties = properties
+    }
+{{/lombok}}
+
+    public Mono<String> send(String message) {
+        return Mono.fromCallable(() -> buildRequest(message))
+                .flatMap(request -> Mono.fromFuture(client.sendMessage(request)))
+                .doOnNext(response -> log.debug("Message sent {}", response.messageId()))
+                .map(SendMessageResponse::messageId);
+    }
+
+    private SendMessageRequest buildRequest(String message) {
+        return SendMessageRequest.builder()
+                .queueUrl(properties.getQueueUrl())
+                .messageBody(message)
+                .build();
+    }
+}

src/main/resources/driven-adapter/sqs/sqs-sender-config.java.mustache

@@ -26,7 +26,7 @@ public class SQSSenderConfig {
     @Bean
     public SqsAsyncClient configSqs(SQSSenderProperties properties, MetricPublisher publisher) {
         return SqsAsyncClient.builder()
-                // .endpointOverride(URI.create("http://localhost:4566")) // TODO: enable only for localstack
+                .endpointOverride(resolveEndpoint(properties))
                 .region(Region.of(properties.getRegion()))
                 .overrideConfiguration(o -> o.addMetricPublisher(publisher))
                 .credentialsProvider(getProviderChain())
@@ -37,7 +37,7 @@ public class SQSSenderConfig {
     @Bean
     public SqsAsyncClient configSqs(SQSSenderProperties properties) {
         return SqsAsyncClient.builder()
-                // .endpointOverride(URI.create("http://localhost:4566")) // TODO: enable only for localstack
+                .endpointOverride(resolveEndpoint(properties))
                 .region(Region.of(properties.getRegion()))
                 .credentialsProvider(getProviderChain())
                 .build();
@@ -54,4 +54,11 @@ public class SQSSenderConfig {
                 .addCredentialsProvider(InstanceProfileCredentialsProvider.create())
                 .build();
     }
+
+    private URI resolveEndpoint(SQSSenderProperties properties) {
+        if (properties.getEndpoint() != null) {
+            return URI.create(properties.getEndpoint());
+        }
+        return null;
+    }
 }

src/main/resources/driven-adapter/sqs/sqs-sender-properties.java.mustache

@@ -14,6 +14,7 @@ import org.springframework.context.annotation.Configuration;
 public class SQSSenderProperties {
     private String region;
     private String queueUrl;
+    private String endpoint;
 {{^lombok}}
 
     public SQSSenderProperties() {
@@ -34,5 +35,13 @@ public class SQSSenderProperties {
     public void setQueueUrl(String queueUrl) {
         this.queueUrl = queueUrl;
     }
+
+    public String getEndpoint() {
+        return endpoint;
+    }
+
+    public void setEndpoint(String endpoint) {
+        this.endpoint = endpoint;
+    }
 {{/lombok}}
 }

src/main/resources/driven-adapter/sqs/sqs-sender.java.mustache

@@ -3,10 +3,10 @@ package {{package}}.sqs.sender;
 import {{package}}.sqs.sender.config.SQSSenderProperties;
 {{#lombok}}
 import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.log4j.Log4j2;
 {{/lombok}}
 import org.springframework.stereotype.Service;
-import reactor.core.publisher.Mono;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
 import software.amazon.awssdk.services.sqs.model.SendMessageResponse;
@@ -28,14 +28,26 @@ public class SQSSender /*implements SomeGateway*/ {
         this.client = client;
         this.properties = properties
     }
+
+    public String send(String message) {
+        try {
+            SendMessageResponse response = client.sendMessage(buildRequest(message)).get();
+            log.debug("Message sent {}", response.messageId());
+            return response.messageId();
+        } catch (ExecutionException | InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
 {{/lombok}}
+{{#lombok}}
 
-    public Mono<String> send(String message) {
-        return Mono.fromCallable(() -> buildRequest(message))
-                .flatMap(request -> Mono.fromFuture(client.sendMessage(request)))
-                .doOnNext(response -> log.debug("Message sent {}", response.messageId()))
-                .map(SendMessageResponse::messageId);
+    @SneakyThrows
+    public String send(String message) {
+        SendMessageResponse response = client.sendMessage(buildRequest(message)).get();
+        log.debug("Message sent {}", response.messageId());
+        return response.messageId();
     }
+{{/lombok}}
 
     private SendMessageRequest buildRequest(String message) {
         return SendMessageRequest.builder()