[AWS] SpringBoot LocalStack으로 SNS, SQS 구성하기

 

 

 

LocalStack이란?

Localstack은 AWS Cloud 리소스들을 로컬에서 테스트할 수 있게 도와주는 테스트/모킹 프레임워크를 제공합니다.

LocalStack을 활용하면 장점

• AWS 클라우드 리소스를 직접 수행하지 않고 로컬 테스트가 가능해진다.

 

준비물

• Docker
• AWS CLI

Docker-Compose.YML

version: '3.4'

services:
  localstack:
    image: localstack/localstack:latest
    container_name: localstack_sns_sqs
    ports:
      - '4566:4566'
    environment:
      - DEFAULT_REGION=ap-northeast-2
      - SERVICES=sns,sqs
      - DEBUG=1
      - DATA_DIR=/tmp/localstack/data
    volumes:
      - '/var/run/docker.sock:/var/run/docker.sock'

우선 sns부터 테스트해보고자 합니다.

 

Docker-Compose 실행

//docker-compose.yml 사용하는 경우
docker-compose up -d

//저는 docker-compose-test 라는 이름을 사용해서 다음과 같이 사용했습니다.
docker-compose -f docker-compose-test.yml up --build -d

docker ps

//docker 내리기
docker-compose down

 

Spring Boot with SNS Integration

gradle.kts

implementation("com.amazonaws:aws-java-sdk-sqs:1.11.970")
implementation("com.amazonaws:aws-java-sdk-sns:1.11.970")

 

AwsSnsConfig

import com.amazonaws.auth.AWSStaticCredentialsProvider
import com.amazonaws.auth.BasicAWSCredentials
import com.amazonaws.client.builder.AwsClientBuilder
import com.amazonaws.services.sns.AmazonSNS
import com.amazonaws.services.sns.AmazonSNSClient
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

@Configuration
class AWSSNSConfig {

  @Bean(destroyMethod = "shutdown")
  fun amazonSNS(): AmazonSNS {
    return AmazonSNSClient.builder()
      .withEndpointConfiguration(
        AwsClientBuilder.EndpointConfiguration(
          "http://localhost:4566", "ap-northeast-2"
        )
      )
      .withCredentials(
        AWSStaticCredentialsProvider(
          BasicAWSCredentials("foo", "bar")
        )
      )
      .build()
  }
}

 

이제 다음과 같은 작업을 수행할 수 있습니다.

• 토픽 만들기
• 토픽 목록
• 메시지 게시
• 토픽 구독하기
• 목록 구독
• 토픽 구독 취소하기
• 토픽 삭제

 

SpringBootTeset로 검증

import com.amazonaws.services.sns.AmazonSNS
import com.amazonaws.services.sns.model.*
import org.junit.jupiter.api.*
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import java.util.*


@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@TestMethodOrder(MethodOrderer.OrderAnnotation::class)
class TestSNS {

  private val topic = "topic"

  @Autowired
  private lateinit var amazonSNS: AmazonSNS

  private lateinit var topicArn: String

  private lateinit var subscriptionArn: String

  @Test
  @Order(1)
  fun testCreateTopic() {
    val createTopic = amazonSNS.createTopic(topic)
    topicArn = createTopic.topicArn
    Assertions.assertEquals(200, createTopic.sdkHttpMetadata.httpStatusCode)
  }

  @Test
  @Order(2)
  fun testListTopics() {
    val request = ListTopicsRequest()
    val result = amazonSNS.listTopics(request)
    Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    Assertions.assertTrue(result.topics.isNotEmpty())
    Assertions.assertTrue(result.topics.contains(Topic().withTopicArn(topicArn)))
  }

  @Test
  @Order(3)
  fun testPublish() {
    val result = amazonSNS.publish(topicArn, "this is a sample message")
    Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    Assertions.assertNotNull(result.messageId)
  }

  @Test
  @Order(4)
  fun testSubscribe() {
    val request = SubscribeRequest()
    request.protocol = "email"
    request.endpoint = "example@turkdogan.dev"
    request.topicArn = topicArn
    val result = amazonSNS.subscribe(request)
    subscriptionArn = result.subscriptionArn
    Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
  }

  @Test
  @Order(5)
  fun testListSubscriptions() {
    val result = amazonSNS.listSubscriptions()
    Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    Assertions.assertTrue(result.subscriptions.isNotEmpty())
    println(result.subscriptions)
    val subscription = result.subscriptions.first()
    Assertions.assertEquals("email", subscription.protocol)
    Assertions.assertEquals("example@turkdogan.dev", subscription.endpoint)
  }

  @Test
  @Order(6)
  fun testUnsubscribe() {
    val request = UnsubscribeRequest()
    request.subscriptionArn = subscriptionArn
    val result = amazonSNS.unsubscribe(request)
    Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
  }

  @Test
  @Order(7)
  fun testDeleteTopic() {
    Assertions.assertDoesNotThrow {
      val result = amazonSNS.deleteTopic(topicArn)
      Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    }
  }
}

 

AWSSQSConfig

import com.amazonaws.auth.AWSStaticCredentialsProvider
import com.amazonaws.auth.BasicAWSCredentials
import com.amazonaws.client.builder.AwsClientBuilder
import com.amazonaws.services.sqs.AmazonSQSAsync
import com.amazonaws.services.sqs.AmazonSQSAsyncClient
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

@Configuration
class AWSSQSConfig {

  @Bean(destroyMethod = "shutdown")
  fun amazonSQS(): AmazonSQSAsync {
    return AmazonSQSAsyncClient.asyncBuilder()
            .withEndpointConfiguration(AwsClientBuilder.EndpointConfiguration(
                    "http://localhost:4566", "us-east-1"))
            .withCredentials(AWSStaticCredentialsProvider(
                    BasicAWSCredentials("foo", "bar")))
            .build()
  }
}

 

SQS Test로 검증

import com.amazonaws.services.sqs.AmazonSQSAsync
import com.amazonaws.services.sqs.model.CreateQueueRequest
import com.amazonaws.services.sqs.model.DeleteMessageRequest
import com.amazonaws.services.sqs.model.Message
import com.amazonaws.services.sqs.model.SendMessageRequest
import org.junit.jupiter.api.*
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import java.util.*

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@TestMethodOrder(MethodOrderer.OrderAnnotation::class)
class TestSQS() {

    private val queue = UUID.randomUUID().toString()

    @Autowired
    private lateinit var amazonSQS: AmazonSQSAsync

    private lateinit var queueUrl: String

    private lateinit var message: Message

    @Test
    @Order(1)
    fun testCreateQueue() {
        val result = amazonSQS.createQueue(queue)
        queueUrl = result.queueUrl
        Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    }

    @Test
    @Order(1)
    fun testCreateFifoQueue() {
        val request = CreateQueueRequest()
        request.queueName = "$queue.fifo"
        request.addAttributesEntry("FifoQueue", "true")
        val result = amazonSQS.createQueue(request)
        Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    }

    @Test
    @Order(2)
    fun testListQueues() {
        val result = amazonSQS.listQueues()
        Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
        Assertions.assertTrue(result.queueUrls.isNotEmpty())
        Assertions.assertTrue(result.queueUrls.contains(queueUrl))
    }

    @Test
    @Order(3)
    fun testSendMessage() {
        val request = SendMessageRequest()
        request.messageBody = "This is SQS message"
        request.queueUrl = queueUrl
        val result = amazonSQS.sendMessage(request)
        val messageId = result.messageId
        Assertions.assertNotNull(messageId)
        val receiveMessageResult = amazonSQS.receiveMessage(queueUrl)
        message = receiveMessageResult.messages.first()
        Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
        Assertions.assertEquals(request.messageBody, message.body)
        Assertions.assertEquals(messageId, message.messageId)
    }

    @Test
    @Order(4)
    fun testDeleteMessage() {
        val request = DeleteMessageRequest()
        request.queueUrl = queueUrl
        request.receiptHandle = message.receiptHandle
        val result = amazonSQS.deleteMessage(request)
        val receiveMessageResult = amazonSQS.receiveMessage(queueUrl)
        Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
        Assertions.assertTrue(receiveMessageResult.messages.isNullOrEmpty())
    }

    @Test
    @Order(5)
    fun testDeleteQueue() {
        val result = amazonSQS.deleteQueue(queueUrl)
        Assertions.assertEquals(200, result.sdkHttpMetadata.httpStatusCode)
    }
}

 

 

Spring Boot Local 실행하여 SNS, SQS 연동되는지 확인하기

gradle.kts

implementation("org.springframework.cloud:spring-cloud-starter-aws-messaging:2.2.6.RELEASE")

 

LocalMessagingConfig

@Configuration
@Profile("local")
class LocalMessagingConfig(
  private val amazonSNS: AmazonSNS,
  private val amazonSQS: AmazonSQS,
) {
  lateinit var topicArn: String
  lateinit var queue: String

  @PostConstruct
  fun generateSnsTopicAndSubscribeSQS() {
    val createTopic = amazonSNS.createTopic("MemberSignUpTopic")
    topicArn = createTopic.topicArn
    println(topicArn)

    val result = amazonSQS.createQueue("MemberSessionSaveQueue")
    queue = result.queueUrl
    println(queue)

    val subscribeQueue = Topics.subscribeQueue(amazonSNS, amazonSQS, topicArn, queue)
    println(subscribeQueue)
  }

}

MemberSignUpTopic이라는 SNS를 생성합니다.

MemberSessionSaveQueue라는 SQS를 생성합니다.

SQS가 SNS를 구독하도록 설정합니다.

 

LocalPublishSignUpEvent

@Component
@Profile("local")
class LocalPublishSignUpEvent(
  private val amazonSNS: AmazonSNS,
  private val localMessagingConfig: LocalMessagingConfig,
) {
  @TransactionalEventListener
  fun awsSnsPublishTest(signUpEvent: SignUpEvent): PublishResult {
    val subject = "TEST 제목"
    return amazonSNS.publish(localMessagingConfig.topicArn, subject, signUpEvent.toString())
  }
}

SignUpEvent가 ApplicationEventPublisher에 의해 발생되면 SNS를 publish 합니다.

 

LocalSqsConsumeSignUpEvent

@Component
@Profile("local")
class LocalSqsConsumeSignUpEvent {

  @SqsListener(value = ["MemberSessionSaveQueue"], deletionPolicy = SqsMessageDeletionPolicy.NEVER)
  fun listen(@Payload message: String, ack: Acknowledgment) {
    println("MemberSessionSaveQueue Message Consume Message: $message")
    ack.acknowledge();
  }
}

@SqsListener를 통하여 SQS에 메시지가 적재되면 해당 메시지를 꺼내옵니다.

 

 

주의사항

실행, 테스트시 다음 설정이 추가되어야 합니다.

 

application.yml 

cloud:
  aws:
    region:
      static: ap-northeast-2
    stack:
      auto: false