package com.lcg.rabbitstable.scopeoperations;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * 使用场景：
 * <p>基于同一个channel发送所有的消息</p>
 * <p>只需要简单的确认信息返回</p>
 * <p></p>
 */
@Slf4j
@Profile("scoped-operations")
@Configuration
@EnableScheduling
public class ScopedOperationConfig {

    protected static final String queue_name = "so.queue.0";
    protected static final String exchange_name = "so.exchange.0";
    @Lazy
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Bean
    public CachingConnectionFactory factory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);

        factory.setUri("amqp://localhost:5672");
        factory.setUsername("guest");
        factory.setPassword("<PASSWORD>");

        factory.addChannelListener((ch, transaction) -> log.info("channel connected:{}, tx:{}", ch.getChannelNumber(), transaction));
        return factory;
    }

    @Bean
    public AmqpAdmin admin() {
        RabbitAdmin admin = new RabbitAdmin(factory());

        Queue queue = QueueBuilder.durable(queue_name).build();
        admin.declareQueue(queue);

        TopicExchange exchange = new TopicExchange(exchange_name);

        admin.declareExchange(exchange);

        admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(""));

        return admin;
    }

    @RabbitListener(queues = queue_name)
    public void handler(String message) {
        log.info("handle massage: {}", message);
    }

    @Scheduled(cron = "0/4 * * * * ?")
    public void sendMsgWithoutCallback() {
        List<String> list = new ArrayList<>();
        list.add("a_" + UUID.randomUUID().toString());
        list.add("b_" + UUID.randomUUID().toString());
        list.add("c_" + UUID.randomUUID().toString());
        Boolean result = this.rabbitTemplate.invoke(t -> {
            list.forEach(str -> t.convertAndSend(exchange_name, "", str));
            t.waitForConfirmsOrDie(5000);
            return true;
        });
        log.info("non-callback result: {}", result);
    }

    @Scheduled(cron = "0/5 * * * * ?")
    public void sendMsgWithCallback() {
        List<String> list = new ArrayList<>();
        list.add("1_" + UUID.randomUUID().toString());
        list.add("2_" + UUID.randomUUID().toString());
        list.add("3_" + UUID.randomUUID().toString());
        list.add("4_" + UUID.randomUUID().toString());
        Boolean result = this.rabbitTemplate.invoke(t -> {
                    list.forEach(str -> t.convertAndSend(exchange_name, "", str));
                    t.waitForConfirmsOrDie(5000);
                    return true;
                },
                (tag, multiple) -> {
                    log.info("callback Ack: " + tag + ":" + multiple);
                }, (tag, multiple) -> {
                    log.info("callback Nack: " + tag + ":" + multiple);
                }
        );
        log.info("callback result: {}", result);
    }
}
