iundarigun

@EnableCaching
@SpringBootApplication
class DistributedCacheApplication

fun main(args: Array<String>) {
	runApplication<DistributedCacheApplication>(*args)
}

iundarigun

@Tag("integration")
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class EmployeesEndpoint {
  // code
}

iundarigun

@Service
class AckListener {
    private val log = LoggerFactory.getLogger(javaClass)

    @RabbitListener(queues = ["FIRST-QUEUE-ADVANCED"], ackMode = "MANUAL")
    fun consumerFirstQueue(person: Person, channel: Channel, @Header(AmqpHeaders.DELIVERY_TAG) tag: Long?) {
        log.info("person $person")
        if (person.collageCompletedYear == null) {
            log.warn("message wrong")
            channel.basicNack(tag ?: 0L, false, false)

iundarigun

@ConstructorBinding
@ConfigurationProperties(prefix = "spring.rabbitmq")
data class ConnectionConfig(
    val host:String,
    val port: Int,
    val username: String,
    val password: String,
    val virtualHost: String,
    val alternativeHost:String,
    val alternativePort: Int,

iundarigun

    private fun retryPolicy(): Advice {
        return RetryInterceptorBuilder
            .stateless()
            .maxAttempts(5)
            .backOffOptions(
                1000, // Initial interval
                2.0, // Multiplier
                6000 // Max interval
            )
            .recoverer(RejectAndDontRequeueRecoverer())

iundarigun

    @Bean
    fun listenerContainer(): MessageListenerContainer {
        val container = SimpleMessageListenerContainer()
        container.connectionFactory = connectionFactory
        container.setQueueNames("SECOND-QUEUE-ADVANCED")
        container.setMessageListener(queueListener)
        simpleRabbitListenerContainerFactory.adviceChain?.let {
            container.setAdviceChain(*it, retryPolicy())
        }
        container.start()

iundarigun

@Configuration
class ConsumerConfig(
    private val connectionFactory: ConnectionFactory,
    private val queueListener: QueueListener,
    private val simpleRabbitListenerContainerFactory: SimpleRabbitListenerContainerFactory
) {

    @Bean
    fun listenerContainer(): MessageListenerContainer {
        val container = SimpleMessageListenerContainer()

iundarigun

    private fun createSecondQueue(rabbitAdmin: RabbitAdmin) {
        val queue = QueueBuilder.durable(QueueDefinition.SECOND_QUEUE)
            .maxLength(10)
            .ttl(30_000)
            .deadLetterExchange(QueueDefinition.DLQ_EXCHANGE)
            .deadLetterRoutingKey(QueueDefinition.DLQ_BINDING_KEY)
            .build()
        val binding = Binding(
            QueueDefinition.SECOND_QUEUE,
            Binding.DestinationType.QUEUE,

iundarigun

@Configuration
class RabbitConfig(private val connectionFactory: ConnectionFactory) {

    @PostConstruct
    fun createRabbitElements() {
        val rabbitAdmin = RabbitAdmin(connectionFactory)
        createExchange(rabbitAdmin)
        createFirstQueue(rabbitAdmin)
    }

iundarigun

@Service
class JsonConsumer (
    private val messageConverter: MessageConverter
){
    private val log = LoggerFactory.getLogger(javaClass)

    @RabbitListener(queues = ["JSON-QUEUE-BASIC"])
    fun receiveMessageFromJsonQueue(message: Message) {
        log.info("receive message from ${message.messageProperties.consumerQueue}")
        val person = messageConverter.fromMessage(message)