package cn.axzo.workflow.starter.handler.execute.interceptor;

import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.workflow.starter.BroadcastListenerProperties;
import cn.axzo.workflow.starter.handler.execute.ListenerExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.function.Consumer;

/**
 * 当前listener执行出错，忽略继续往下执行，可配置重试相关参数，不抛出异常
 */
public final class FailOverInterceptor extends AbstractListenerInterceptor {
    private static final Logger log = LoggerFactory.getLogger(FailOverInterceptor.class);

    private int numOfRetries = 3;
    private int waitTimeInMs = 1500;
    private int waitIncreaseFactor = 3;

    public FailOverInterceptor() {
    }

    public FailOverInterceptor(BroadcastListenerProperties broadcast) {
        this.numOfRetries = 0 == Math.abs(broadcast.getNumOfRetries()) ? this.numOfRetries : Math.abs(broadcast.getNumOfRetries());
        this.waitTimeInMs = Math.abs(broadcast.getWaitTimeInMs());
        this.waitIncreaseFactor = Math.abs(broadcast.getWaitIncreaseFactor());
    }

    @Override
    public <T> void execute(ListenerExecutor executor, Consumer<T> consumer, EventConsumer.Context context, T t) {
        // 避免组件抛出“程序处理超时”的异常信息, 为业务冗余 10 处理时间
        context.setMaxAllowElapsedMillis(calc() + 10_000L);
        long waitTime = waitTimeInMs;
        int failedAttempts = 0;
        do {
            if (failedAttempts > 0) {
                if (log.isDebugEnabled()) {
                    log.debug("Waiting for {} ms before retrying the command. retryTimes: {}", waitTime, failedAttempts);
                }
                waitBeforeRetry(waitTime);
                waitTime *= waitIncreaseFactor;
            }
            try {
                getNext().execute(executor, consumer, context, t);
                return;
            } catch (Throwable e) {
                if (failedAttempts == numOfRetries) {
                    log.error("Workflow Engine Starter caught exception: {}", e.getMessage(), e);
                    throw e;
                }
            }
            failedAttempts++;
        } while (failedAttempts <= numOfRetries);
    }

    private Long calc() {
        long waitTime = waitTimeInMs;
        for (int failedAttempts = 0; failedAttempts < numOfRetries; failedAttempts++) {
            // 更新等待时间
            waitTime *= waitIncreaseFactor;
        }
        // 返回最后一次等待时间（注意：这可能不是最终你想要的逻辑，取决于你的具体需求）
        return waitTime;
    }

    private void waitBeforeRetry(long waitTime) {
        try {
            Thread.sleep(waitTime);
        } catch (InterruptedException e) {
            log.debug(" interrupted while waiting for a retry.");
        }
    }
}
