Skip to content

Instantly share code, notes, and snippets.

@wrschneider
Last active April 22, 2023 16:27
Show Gist options
  • Save wrschneider/42407cc2ea70799362cc5b044ebcfabb to your computer and use it in GitHub Desktop.
Save wrschneider/42407cc2ea70799362cc5b044ebcfabb to your computer and use it in GitHub Desktop.
Spring Boot listener for Amazon SQS
package bill.boottest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
@SpringBootApplication
public class App implements CommandLineRunner {
private static Logger LOG = LoggerFactory.getLogger(App.class);
public static void main(String[] args) {
SpringApplication.run(App.class, args);
}
@SqsListener("bill-poc")
public void listen(DataObject message) {
LOG.info("!!!! received message {} {}", message.getFoo(), message.getBar());
}
@Override
public void run(String... args) throws Exception {
}
public static class DataObject {
private String foo;
private String bar;
@JsonCreator
public DataObject(@JsonProperty("foo") String foo, @JsonProperty("bar") String bar) {
this.foo = foo;
this.bar = bar;
}
public String getFoo() {
return foo;
}
public String getBar() {
return bar;
}
}
}
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>bill</groupId>
<artifactId>boottest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>boottest</name>
<url>http://maven.apache.org</url>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.5.RELEASE</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-aws</artifactId>
<version>2.0.0.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-aws</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-aws-messaging</artifactId>
</dependency>
</dependencies>
</project>
@mengjiann
Copy link

Just a minor comment for those who bumped into this, if you are using spring-cloud-aws-messaging as standalone dependency without spring-cloud-starter-aws. My use case is running it locally using mock sqs library: https://github.com/softwaremill/elasticmq.

This is what you will need to configure the AmazonSQSAsync client:

@Configuration
public class AwsSqsConfig {

    @Value("${app.aws.sqs.mock.endpoint}")
    private String awsSqsMockEndpoint;

    @Bean
    @Primary
    public AmazonSQSAsync awsSqsClientMock(){
        return AmazonSQSAsyncClientBuilder.standard()
                .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("x", "x")))
                .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(awsSqsMockEndpoint, "elasticmq"))
                .build();
    }

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(AmazonSQSAsync amazonSQSAsync, QueueMessageHandler queueMessageHandler) {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
        simpleMessageListenerContainer.setAmazonSqs(amazonSQSAsync);
        simpleMessageListenerContainer.setMessageHandler(queueMessageHandler);
        simpleMessageListenerContainer.setMaxNumberOfMessages(10);
        simpleMessageListenerContainer.setTaskExecutor(threadPoolTaskExecutor());
        return simpleMessageListenerContainer;
    }

    @Bean
    public QueueMessageHandler queueMessageHandler(AmazonSQSAsync amazonSQSAsync) {
        QueueMessageHandlerFactory queueMessageHandlerFactory = new QueueMessageHandlerFactory();
        queueMessageHandlerFactory.setAmazonSqs(amazonSQSAsync);
        QueueMessageHandler queueMessageHandler = queueMessageHandlerFactory.createQueueMessageHandler();
        return queueMessageHandler;
    }


    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(10);
        executor.initialize();
        return executor;
    }

}

@JoniSykes
Copy link

@mengjiann I'm using alpine-sqs in docker locally, and trying to connect to it using these (and similar) examples.
If I use the config you provided above, does that mean I do not need a .aws/credentials or config file locally, as it relies on the values supplied in the config class?
Also, my alpine-sqs runs on default port on localhost over http (not https), but the SDK is complaining about plaintext connection. Any ideas? Anything you can share from your use case?

Thanks in advanced.
Jonny

@lmyslinski
Copy link

@JoniSykes for what it's worth now - yes, this config means it's not using any of the local files, all config is configured here. I'm running alpine-sqs as well and have not had the problem with https. I also found a simpler way to get it working, if you don't need explicit control over thread pool here:

@EnableSqs
@Configuration
public class SqsConfig {
    String endpoint = "http://localhost:9324";
    String region = "elasticmq";

    private final AmazonSQSAsync sqsClient = AmazonSQSAsyncClientBuilder.standard()
            .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, region))
            .build();

    @Bean
    public QueueMessagingTemplate getMessageTemplate(){
        return new QueueMessagingTemplate(sqsClient);
    }

    @Bean
    @Primary
    public AmazonSQSAsync amazonSQSAsync(){
        return sqsClient;
    }
}

I don't why any other tutorial doesn't mention it, but if I don't use @EnableSqs the @SqsListener annotation does not work and I'm not getting any messages.

@sovietspy2
Copy link

@EnableSqs

For me it's working without the annotation if I define the SimpleMessageListenerContainer and QueueMessageHandler beans. This gist is a great resource to get it up and running thanks yall!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment