package study.u47;

import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.retry.RetryPolicy;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * U47 Spring Integration：通道语义、变换、重试与错误通道的可恢复消息流。
 * 纯进程内实验：无外部服务与持久化资源，上下文关闭即结束。
 */
public class U47Lab {

    static final AtomicInteger RETRY_ATTEMPTS = new AtomicInteger();

    public static void main(String[] args) throws Exception {
        try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(FlowConfig.class)) {
            // ---- 1. DirectChannel：同线程执行，异常直接抛给发送方（不进错误通道） ----
            DirectChannel direct = context.getBean("direct", DirectChannel.class);
            final String[] handlerThread = {null};
            direct.subscribe(msg -> {
                handlerThread[0] = Thread.currentThread().getName();
                throw new IllegalStateException("boom-direct");
            });
            String senderThread = Thread.currentThread().getName();
            boolean wrappedToCaller;
            try {
                direct.send(MessageBuilder.withPayload("x").build());
                wrappedToCaller = false;
                throw new AssertionError("DirectChannel 应向发送方抛出异常");
            } catch (MessagingException expected) {
                // 发送方线程收到包装后的 MessageDeliveryException，cause 是处理器的原始异常
                wrappedToCaller = expected.getFailedMessage() != null
                        && expected.getCause() instanceof IllegalStateException;
            }
            boolean sameThread = senderThread.equals(handlerThread[0]);
            check(wrappedToCaller && sameThread, "DirectChannel 应在同线程把（包装后的）异常抛给发送方");
            out("directChannel exceptionToCaller=true wrappedAsMessageDeliveryException=true sameThread=" + sameThread);

            // ---- 2. 变换端点：@Transformer 在通道间改写 payload ----
            MessageChannel csvChannel = context.getBean("csvChannel", MessageChannel.class);
            PollableChannel csvResult = context.getBean("csvResult", PollableChannel.class);
            csvChannel.send(MessageBuilder.withPayload("a,b,c").build());
            Message<?> transformed = csvResult.receive(5_000);
            check(transformed != null && transformed.getPayload().equals(List.of("a", "b", "c")),
                    "Transformer 应拆分为三元素，实际 " + (transformed == null ? "null" : transformed.getPayload()));
            out("transformer payloadTransformed=true size=3");

            // ---- 3. QueueChannel + 轮询端点：异步失败按 errorChannel 头进指定通道 ----
            PollableChannel workQueue = context.getBean("workQueue", PollableChannel.class);
            PollableChannel errorQueue = context.getBean("errorQueue", PollableChannel.class);
            workQueue.send(MessageBuilder.withPayload("poison").setErrorChannel(errorQueue).build());
            Message<?> errorMessage = errorQueue.receive(10_000);
            check(errorMessage instanceof ErrorMessage, "轮询端点失败应产生 ErrorMessage");
            Throwable errorPayload = ((ErrorMessage) errorMessage).getPayload();
            check(errorPayload instanceof MessagingException
                    && "poison".equals(((MessagingException) errorPayload).getFailedMessage().getPayload()),
                    "ErrorMessage 应携带 failedMessage");
            out("queueChannel asyncError=true errorMessageOnCustomChannel=true failedPayload=poison");

            // ---- 4. 重试：前 2 次失败第 3 次成功，观察本组合的重试次数语义 ----
            MessageChannel retryChannel = context.getBean("retryChannel", MessageChannel.class);
            PollableChannel retryResult = context.getBean("retryResult", PollableChannel.class);
            RETRY_ATTEMPTS.set(0);
            retryChannel.send(MessageBuilder.withPayload("retry-me").build());
            Message<?> retried = retryResult.receive(10_000);
            check(retried != null && "recovered:retry-me".equals(retried.getPayload()), "重试后应成功返回");
            out("retry attempts=" + RETRY_ATTEMPTS.get() + " recoveredOn=" + RETRY_ATTEMPTS.get());

            // ---- 5. 重试耗尽 + ErrorMessageSendingRecoverer：错误进恢复通道，主流程修复后重新投递 ----
            MessageChannel exhaustChannel = context.getBean("exhaustChannel", MessageChannel.class);
            PollableChannel exhaustError = context.getBean("exhaustError", PollableChannel.class);
            PollableChannel exhaustResult = context.getBean("exhaustResult", PollableChannel.class);
            exhaustChannel.send(MessageBuilder.withPayload("always-bad").build());
            Message<?> recovered = exhaustError.receive(10_000);
            check(recovered instanceof ErrorMessage, "重试耗尽应经 recovery 回调进错误队列");
            Message<?> failedMessage = ((MessagingException) ((ErrorMessage) recovered).getPayload()).getFailedMessage();
            check("always-bad".equals(failedMessage.getPayload()), "恢复消息应携带原始 failedMessage");
            exhaustChannel.send(MessageBuilder.withPayload("fixed").build());
            Message<?> ok = exhaustResult.receive(10_000);
            check(ok != null && "ok:fixed".equals(ok.getPayload()), "修复后重新投递应成功");
            out("retryExhausted recoveryToErrorChannel=true redelivered=true finalResult=ok:fixed");

            // ---- 6. ExpressionEvaluatingRequestHandlerAdvice：失败表达式路由 + trapException ----
            MessageChannel evalChannel = context.getBean("evalChannel", MessageChannel.class);
            PollableChannel evalFailure = context.getBean("evalFailure", PollableChannel.class);
            boolean evalThrew;
            try {
                evalChannel.send(MessageBuilder.withPayload("bad").build());
                evalThrew = false;
            } catch (RuntimeException expected) {
                evalThrew = true;
            }
            Message<?> failureNote = evalFailure.receive(5_000);
            check(failureNote instanceof ErrorMessage, "failureChannel 应收到 ErrorMessage");
            Throwable notePayload = ((ErrorMessage) failureNote).getPayload();
            check(notePayload instanceof ExpressionEvaluatingRequestHandlerAdvice.MessageHandlingExpressionEvaluatingAdviceException,
                    "错误载荷应是 advice 异常包装，实际 " + notePayload.getClass().getSimpleName());
            Object evalResult = ((ExpressionEvaluatingRequestHandlerAdvice.MessageHandlingExpressionEvaluatingAdviceException) notePayload)
                    .getEvaluationResult();
            check("recorded:bad".equals(evalResult), "表达式结果应在异常包装内，实际 " + evalResult);
            check(!evalThrew, "trapException=true 时异常不应穿透到调用方");
            out("advice onFailureRouted=true trapped=true recorded=recorded:bad");
        }
    }

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

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

    @Configuration
    @EnableIntegration
    static class FlowConfig {

        @Bean
        public DirectChannel direct() {
            return new DirectChannel();
        }

        @Bean
        public DirectChannel csvChannel() {
            return new DirectChannel();
        }

        @Bean
        public QueueChannel csvResult() {
            return new QueueChannel(10);
        }

        @Transformer(inputChannel = "csvChannel", outputChannel = "csvResult")
        public List<String> transform(String payload) {
            return List.of(payload.split(","));
        }

        @Bean
        public QueueChannel workQueue() {
            return new QueueChannel(10);
        }

        @Bean
        public QueueChannel errorQueue() {
            return new QueueChannel(10);
        }

        @Bean
        @ServiceActivator(inputChannel = "workQueue", poller = @Poller(fixedDelay = "50"))
        public MessageHandler poisonHandler() {
            return msg -> {
                if ("poison".equals(msg.getPayload())) {
                    throw new MessagingException(msg, "poison payload");
                }
            };
        }

        @Bean
        public DirectChannel retryChannel() {
            return new DirectChannel();
        }

        @Bean
        public QueueChannel retryResult() {
            return new QueueChannel(10);
        }

        @Bean
        public RequestHandlerRetryAdvice retryAdvice() {
            RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
            advice.setRetryPolicy(RetryPolicy.builder()
                    .maxRetries(2)
                    .delay(Duration.ofMillis(5))
                    .build());
            return advice;
        }

        @ServiceActivator(inputChannel = "retryChannel", outputChannel = "retryResult", adviceChain = "retryAdvice")
        public String retryHandler(String payload) {
            if (RETRY_ATTEMPTS.incrementAndGet() < 3) {
                throw new IllegalStateException("transient");
            }
            return "recovered:" + payload;
        }

        @Bean
        public DirectChannel exhaustChannel() {
            return new DirectChannel();
        }

        @Bean
        public QueueChannel exhaustResult() {
            return new QueueChannel(10);
        }

        @Bean
        public QueueChannel exhaustError() {
            return new QueueChannel(10);
        }

        @Bean
        public RequestHandlerRetryAdvice exhaustAdvice() {
            RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
            advice.setRetryPolicy(RetryPolicy.builder()
                    .maxRetries(1)
                    .delay(Duration.ofMillis(5))
                    .build());
            advice.setRecoveryCallback(new ErrorMessageSendingRecoverer(exhaustError()));
            return advice;
        }

        @ServiceActivator(inputChannel = "exhaustChannel", outputChannel = "exhaustResult", adviceChain = "exhaustAdvice")
        public String exhaustHandler(String payload) {
            if ("always-bad".equals(payload)) {
                throw new IllegalStateException("permanent failure");
            }
            return "ok:" + payload;
        }

        @Bean
        public DirectChannel evalChannel() {
            return new DirectChannel();
        }

        @Bean
        public QueueChannel evalFailure() {
            return new QueueChannel(10);
        }

        @Bean
        public ExpressionEvaluatingRequestHandlerAdvice evalAdvice() {
            ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
            advice.setOnFailureExpressionString("'recorded:' + payload");
            advice.setFailureChannel(evalFailure());
            advice.setTrapException(true);
            return advice;
        }

        @ServiceActivator(inputChannel = "evalChannel", adviceChain = "evalAdvice")
        public String evalHandler(String payload) {
            throw new IllegalStateException("always fails");
        }
    }
}