package study.u48;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.Duration;

import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * U48 Spring AMQP：启用配置、监听生命周期、确认/异常/重试，真实 AMQP 0-9-1 broker。
 * broker 为 Qpid Broker-J 内存替身（与 rabbit 客户端兼容，非 RabbitMQ 本体）。
 */
public class AmqpLab {

    static final List<String> CONSUMED = new CopyOnWriteArrayList<>();
    static final AtomicInteger RETRY_ATTEMPTS = new AtomicInteger();

    public static void main(String[] args) throws Exception {
        final int port = Integer.parseInt(args[0]);
        out("amqpBroker started=amqp091 port=" + port + " broker=qpid-substitute");

        // 等待 broker 端虚拟主机激活：端口先于 vhost 就绪时会 404，按 500ms 重试至多 30 秒
        AnnotationConfigApplicationContext context = null;
        org.springframework.amqp.AmqpIOException lastFailure = null;
        for (int attempt = 0; attempt < 60 && context == null; attempt++) {
            try {
                context = new AnnotationConfigApplicationContext();
                context.getBeanFactory().registerSingleton("amqpPort", port);
                context.register(AmqpConfig.class);
                context.refresh();
            } catch (org.springframework.amqp.AmqpIOException vhostRace) {
                lastFailure = vhostRace;
                if (context != null) {
                    try {
                        context.close();
                    } catch (RuntimeException ignored) {
                    }
                    context = null;
                }
                Thread.sleep(500);
            }
        }
        if (context == null) {
            throw lastFailure;
        }
        final AnnotationConfigApplicationContext activeContext = context;
        try (activeContext) {

            RabbitTemplate template = activeContext.getBean(RabbitTemplate.class);
            RabbitAdmin admin = activeContext.getBean(RabbitAdmin.class);

            // ---- 1. 启用配置：RabbitAdmin 声明交换机/队列/绑定 ----
            admin.declareExchange(new DirectExchange("u48.ex"));
            admin.declareQueue(new Queue("u48.work", true));
            admin.declareQueue(new Queue("u48.dlq", true));
            admin.declareBinding(BindingBuilder.bind(new Queue("u48.work", true))
                    .to(new DirectExchange("u48.ex")).with("work"));
            admin.declareQueue(new Queue("u48.retry", true));
            admin.declareBinding(BindingBuilder.bind(new Queue("u48.retry", true))
                    .to(new DirectExchange("u48.ex")).with("retry"));
            Integer queueCount = template.execute((org.springframework.amqp.rabbit.core.ChannelCallback<Integer>) ch -> ch.queueDeclarePassive("u48.work").getMessageCount());
            check(queueCount != null && queueCount == 0, "work 队列应已声明");
            out("rabbitDeclared exchange=1 queues=3 bindings=2");

            // 声明完成后再启动监听容器（工厂 autoStartup=false，避免声明竞态）
            MessageListenerContainer container =
                    activeContext.getBean(RabbitListenerEndpointRegistry.class).getListenerContainer("workListener");
            MessageListenerContainer retryContainer =
                    activeContext.getBean(RabbitListenerEndpointRegistry.class).getListenerContainer("retryListener");
            container.start();
            retryContainer.start();

            // ---- 2. 监听生命周期：发送后消费；stop 后积压；start 后恢复 ----
            template.convertAndSend("u48.ex", "work", "msg-1");
            long deadline = System.currentTimeMillis() + 10_000;
            while (CONSUMED.stream().noneMatch(m -> m.contains("msg-1")) && System.currentTimeMillis() < deadline) {
                Thread.sleep(100);
            }
            check(CONSUMED.stream().anyMatch(m -> m.contains("msg-1")), "监听器应消费 msg-1");
            container.stop();
            check(!container.isRunning(), "容器 stop 后应停止");
            template.convertAndSend("u48.ex", "work", "msg-2");
            Thread.sleep(500);
            Integer backlog = template.execute((org.springframework.amqp.rabbit.core.ChannelCallback<Integer>) ch -> ch.queueDeclarePassive("u48.work").getMessageCount());
            check(backlog != null && backlog == 1, "停止期间消息应积压，实际 " + backlog);
            container.start();
            deadline = System.currentTimeMillis() + 10_000;
            while (CONSUMED.stream().noneMatch(m -> m.contains("msg-2")) && System.currentTimeMillis() < deadline) {
                Thread.sleep(100);
            }
            check(CONSUMED.stream().anyMatch(m -> m.contains("msg-2")), "容器 start 后应恢复消费");
            out("rabbitLifecycle consumed=2 stoppedBacklog=1 resumed=true");

            // ---- 3. 异常与死信：defaultRequeueRejected=false 时毒消息进 DLQ ----
            template.convertAndSend("u48.ex", "work", "poison");
            deadline = System.currentTimeMillis() + 10_000;
            Integer dlq = null;
            while (System.currentTimeMillis() < deadline) {
                dlq = template.execute((org.springframework.amqp.rabbit.core.ChannelCallback<Integer>) ch -> ch.queueDeclarePassive("u48.dlq").getMessageCount());
                if (dlq != null && dlq == 1) break;
                Thread.sleep(100);
            }
            check(dlq != null && dlq == 1, "毒消息应死信到 u48.dlq");
            out("rabbitDlq poisonDeadLettered=1 defaultRequeueRejected=false");

            // ---- 4. 重试：无状态 RetryOperationsInterceptor 三次尝试后进恢复 ----
            template.convertAndSend("u48.ex", "retry", "retry-me");
            deadline = System.currentTimeMillis() + 15_000;
            Integer retryDlq = null;
            while (System.currentTimeMillis() < deadline) {
                retryDlq = template.execute((org.springframework.amqp.rabbit.core.ChannelCallback<Integer>) ch -> ch.queueDeclarePassive("u48.dlq").getMessageCount());
                if (retryDlq != null && retryDlq >= 2) break;
                Thread.sleep(100);
            }
            check(RETRY_ATTEMPTS.get() == 3, "重试拦截器应尝试 3 次，实际 " + RETRY_ATTEMPTS.get());
            out("rabbitRetry attempts=3 recoveredTo=dlq");

            containerStopQuietly(container);
            containerStopQuietly(retryContainer);
        }
        out("cleanup listenersStopped=true");
    }

    private static void containerStopQuietly(MessageListenerContainer container) {
        try {
            container.stop();
        } catch (RuntimeException ignored) {
        }
    }

    static void check(boolean condition, String message) {
        if (!condition) throw new AssertionError(message);
    }

    static void out(String line) {
        System.out.println("U48 " + line);
    }

    @Configuration
    @EnableRabbit
    static class AmqpConfig {

        @Bean
        public ConnectionFactory connectionFactory(Integer amqpPort) {
            CachingConnectionFactory factory = new CachingConnectionFactory("127.0.0.1", amqpPort);
            // broker 只提供 ANONYMOUS 机制（Qpid Anonymous 提供者）；rabbit 客户端侧注册对应 SASL
            factory.getRabbitConnectionFactory().setSaslConfig(mechanisms -> new com.rabbitmq.client.SaslMechanism() {
                @Override
                public String getName() {
                    return "ANONYMOUS";
                }

                @Override
                public com.rabbitmq.client.LongString handleChallenge(
                        com.rabbitmq.client.LongString challenge, String username, String password) {
                    return com.rabbitmq.client.impl.LongStringHelper.asLongString("u48");
                }
            });
            return factory;
        }

        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
            return new RabbitAdmin(connectionFactory);
        }

        @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
            RabbitTemplate template = new RabbitTemplate(connectionFactory);
            template.setMandatory(false);
            return template;
        }

        @Bean
        public SimpleRabbitListenerContainerFactory workFactory(ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setDefaultRequeueRejected(false);
            factory.setAutoStartup(false);
            // Qpid 不支持 rabbit 的 x-dead-letter-exchange 扩展；应用侧死信用 RepublishMessageRecoverer
            factory.setAdviceChain(RetryInterceptorBuilder.stateless()
                    .configureRetryPolicy(policy -> policy.maxRetries(0))
                    .recoverer(new RepublishMessageRecoverer(
                            new RabbitTemplate(connectionFactory), "u48.dlq"))
                    .build());
            return factory;
        }

        @Bean
        public SimpleRabbitListenerContainerFactory retryFactory(ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setDefaultRequeueRejected(false);
            factory.setAutoStartup(false);
            // spring-rabbit 4.0 的重试拦截器已切换到 Framework 7 RetryPolicy
            factory.setAdviceChain(RetryInterceptorBuilder.stateless()
                    .configureRetryPolicy(policy -> policy.maxRetries(2).delay(Duration.ofMillis(50)))
                    .recoverer(new RepublishMessageRecoverer(
                            new RabbitTemplate(connectionFactory), "u48.dlq"))
                    .build());
            return factory;
        }

        @RabbitListener(id = "workListener", queues = "u48.work", containerFactory = "workFactory")
        public void work(String payload, Channel channel) {
            if ("poison".equals(payload)) {
                throw new IllegalStateException("poison payload");
            }
            CONSUMED.add(payload + "@" + Thread.currentThread().getName());
        }

        @RabbitListener(id = "retryListener", queues = "u48.retry", containerFactory = "retryFactory")
        public void retry(String payload) {
            if ("retry-me".equals(payload)) {
                RETRY_ATTEMPTS.incrementAndGet();
                throw new IllegalStateException("always fails once per attempt");
            }
            CONSUMED.add(payload);
        }
    }
}