<|start_filename|>sqs-starter/src/test/java/io/reflectoring/sqs/internal/TestMessageHandlerRegistration.java<|end_filename|>
package io.reflectoring.sqs.internal;

import com.amazonaws.services.sqs.AmazonSQS;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.reflectoring.sqs.api.*;

class TestMessageHandlerRegistration implements SqsMessageHandlerRegistration<TestMessage> {

    private final AmazonSQS client;
    private final ObjectMapper objectMapper;
    private final TestMessageHandler messageHandler;

    public TestMessageHandlerRegistration(
            AmazonSQS client,
            ObjectMapper objectMapper,
            TestMessageHandler messageHandler) {
        this.client = client;
        this.objectMapper = objectMapper;
        this.messageHandler = messageHandler;
    }

    @Override
    public SqsMessageHandler<TestMessage> messageHandler() {
        return this.messageHandler;
    }

    @Override
    public String name() {
        return "testMessageHandler";
    }

    @Override
    public SqsMessageHandlerProperties messageHandlerProperties() {
        return new SqsMessageHandlerProperties();
    }

    @Override
    public SqsMessagePollerProperties messagePollerProperties() {
        return new SqsMessagePollerProperties("http://localhost:4576/queue/testMessages");
    }

    @Override
    public AmazonSQS sqsClient() {
        return this.client;
    }

    @Override
    public ObjectMapper objectMapper() {
        return this.objectMapper;
    }
}


<|start_filename|>sqs-starter/src/main/java/io/reflectoring/sqs/internal/SqsAutoConfiguration.java<|end_filename|>
package io.reflectoring.sqs.internal;

import io.reflectoring.sqs.api.SqsMessageHandlerRegistration;
import java.util.List;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
class SqsAutoConfiguration {

  @Bean
  SqsMessageHandlerRegistry sqsMessageHandlerRegistry(List<SqsMessageHandlerRegistration<?>> registrations) {
    return new SqsMessageHandlerRegistry(registrations);
  }

  @Bean
  SqsAutoConfigurationLifecycle sqsLifecycle(SqsMessageHandlerRegistry registry) {
    return new SqsAutoConfigurationLifecycle(registry);
  }

}


<|start_filename|>sqs-starter/src/test/java/io/reflectoring/sqs/internal/TestMessageHandler.java<|end_filename|>
package io.reflectoring.sqs.internal;

import io.reflectoring.sqs.api.SqsMessageHandler;

import java.util.concurrent.atomic.AtomicInteger;

class TestMessageHandler implements SqsMessageHandler<TestMessage> {

    private AtomicInteger counter = new AtomicInteger();

    @Override
    public void handle(TestMessage message) {
        counter.incrementAndGet();
    }

    @Override
    public Class<TestMessage> messageType() {
        return TestMessage.class;
    }

    int getCount() {
        return counter.get();
    }
}


<|start_filename|>sqs-starter/src/main/java/io/reflectoring/sqs/api/DefaultSqsMessageHandlerRegistration.java<|end_filename|>
package io.reflectoring.sqs.api;

import com.fasterxml.jackson.databind.ObjectMapper;

public abstract class DefaultSqsMessageHandlerRegistration<T> implements SqsMessageHandlerRegistration<T> {

    @Override
    public String name() {
        return getClass().getSimpleName();
    }

    @Override
    public SqsMessageHandlerProperties messageHandlerProperties() {
        return new SqsMessageHandlerProperties();
    }

    @Override
    public ObjectMapper objectMapper() {
        return new ObjectMapper();
    }
}


<|start_filename|>sqs-starter-test/src/test/java/io/reflectoring/sqs/SqsTestAnnotationTest.java<|end_filename|>
package io.reflectoring.sqs;

import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.ReceiveMessageResult;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;

@SqsTest(queueNames = {"queue1", "queue2"})
@Tag("IgnoreInCi")
class SqsTestAnnotationTest {

    @Test
    void canInteractWithQueues() {
        AmazonSQS sqsClient = sqsClient();

        assertSendAndReceive(sqsClient, "queue1");
        assertSendAndReceive(sqsClient, "queue2");
    }

    private void assertSendAndReceive(AmazonSQS sqsClient, String queueName) {
        String queueUrl = String.format("http://localhost:4576/queue/%s", queueName);
        SendMessageResult sendMessageResult = sqsClient.sendMessage(new SendMessageRequest(
                queueUrl,
                "message"));
        assertThat(sendMessageResult.getSdkHttpMetadata().getHttpStatusCode()).isEqualTo(200);

        ReceiveMessageResult receiveMessageResult = sqsClient.receiveMessage(queueUrl);
        assertThat(receiveMessageResult.getSdkHttpMetadata().getHttpStatusCode()).isEqualTo(200);
        assertThat(receiveMessageResult.getMessages()).hasSize(1);
    }

    private AmazonSQS sqsClient() {
        return AmazonSQSClientBuilder.standard()
                .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
                        "http://localhost:4576",
                        "us-east"
                )).build();
    }

}


<|start_filename|>sqs-starter/src/main/java/io/reflectoring/sqs/api/SqsMessageHandlerRegistration.java<|end_filename|>
package io.reflectoring.sqs.api;

import com.amazonaws.services.sqs.AmazonSQS;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Include a bean of this type in the Spring application context to register a poller that polls messages
 * from an SQS queue and passes them to an {@link SqsMessageHandler}.
 *
 * @param <T> the type of the messages to process.
 */
public interface SqsMessageHandlerRegistration<T> {

    /**
     * The message handler that shall process the messages polled from SQS.
     */
    SqsMessageHandler<T> messageHandler();

    /**
     * A human-readable name for the message handler. This is used to name the message handler threads.
     */
    String name();

    /**
     * Configuration properties for the message handler.
     */
    SqsMessageHandlerProperties messageHandlerProperties();

    /**
     * Configuration properties for the message poller.
     */
    SqsMessagePollerProperties messagePollerProperties();

    /**
     * The SQS client to use for polling messages from SQS.
     */
    AmazonSQS sqsClient();

    /**
     * The {@link ObjectMapper} to use for deserializing messages from SQS.
     */
    ObjectMapper objectMapper();
}


<|start_filename|>sqs-starter/src/main/java/io/reflectoring/sqs/api/SqsQueueInitializer.java<|end_filename|>
package io.reflectoring.sqs.api;

import com.amazonaws.services.sqs.AmazonSQS;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.InitializingBean;

/**
 * Creates an SQS queue on startup. Useful in tests.
 */
@RequiredArgsConstructor
public class SqsQueueInitializer implements InitializingBean {

    private final AmazonSQS sqsClient;
    private final String queueName;

    private void initializeQueue(AmazonSQS sqsClient, String queueName) {
        sqsClient.createQueue(queueName);
    }

    @Override
    public void afterPropertiesSet() {
        initializeQueue(sqsClient, queueName);
    }
}


<|start_filename|>sqs-starter/src/test/java/io/reflectoring/sqs/internal/TestMessageHandlerConfiguration.java<|end_filename|>
package io.reflectoring.sqs.internal;

import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
class TestMessageHandlerConfiguration {

    @Bean
    AmazonSQS sqsClient() {
        AmazonSQSClientBuilder builder = AmazonSQSClientBuilder.standard();
        builder.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
                "http://localhost:4576",
                "us-east"
        ));
        return builder.build();
    }

    @Bean
    TestMessageHandler messageHandler(){
        return new TestMessageHandler();
    }

    @Bean
    TestMessageHandlerRegistration testMessageHandlerRegistration(AmazonSQS sqsClient, ObjectMapper objectMapper, TestMessageHandler messageHandler) {
        return new TestMessageHandlerRegistration(sqsClient, objectMapper, messageHandler);
    }

    @Bean
    TestMessagePublisher testMessagePublisher(AmazonSQS sqsClient, ObjectMapper objectMapper) {
        return new TestMessagePublisher(sqsClient, objectMapper);
    }

    @Bean
    ObjectMapper objectMapper(){
        return new ObjectMapper();
    }

}


<|start_filename|>sqs-starter/src/main/java/io/reflectoring/sqs/internal/ThreadPools.java<|end_filename|>
package io.reflectoring.sqs.internal;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

class ThreadPools {

    public static ThreadPoolExecutor blockingThreadPool(int threads, int queueSize, String poolName) {
        return new ThreadPoolExecutor(
                threads,
                threads,
                0L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(queueSize),
                new NamedThreadFactory(poolName),
                retryPolicy()
        );
    }

    public static ScheduledThreadPoolExecutor blockingScheduledThreadPool(int threads, String poolName) {
        return new ScheduledThreadPoolExecutor(
                threads,
                new NamedThreadFactory(poolName),
                retryPolicy());
    }

    /**
     * Re-Queues a rejected {@link Runnable} into the thread pool's blocking queue, making the submitting thread wait
     * until the threadpool has capacity again.
     */
    static RejectedExecutionHandler retryPolicy() {
        return (r, executor) -> {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        };
    }

}


<|start_filename|>sqs-starter/src/test/java/io/reflectoring/sqs/internal/TestMessagePublisher.java<|end_filename|>
package io.reflectoring.sqs.internal;

import com.amazonaws.services.sqs.AmazonSQS;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.reflectoring.sqs.api.SqsMessagePublisher;

class TestMessagePublisher extends SqsMessagePublisher<TestMessage> {

    TestMessagePublisher(AmazonSQS sqsClient, ObjectMapper objectMapper) {
        super("http://localhost:4576/queue/testMessages", sqsClient, objectMapper);
    }

}
