Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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

This comment has been minimized.

Copy link

commented May 22, 2019

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

This comment has been minimized.

Copy link

commented May 29, 2019

@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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.