package com.sonic.common.event;

import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;

import org.springframework.beans.BeanUtils;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import lombok.extern.slf4j.Slf4j;

/**
 * 默认EventProducer，抽象里事件的组装和事务后发送的逻辑，提供一个Sender，具体服务具体提供
 */
@Slf4j
public abstract class AbstractEventProducer<M> implements EventProducer<M> {

    private final AfterCommitExecutorImpl afterCommitExecutor;

    private final TaskExecutor taskExecutor;

    /**
     * @return 返回发送消息的发送器
     */
    public abstract BiConsumer<Event, M> getSender();

    private M defaultMeta;

    public AbstractEventProducer(M defaultMeta, TaskExecutor taskExecutor) {
        this.afterCommitExecutor = new AfterCommitExecutorImpl();
        this.taskExecutor = taskExecutor;
        this.defaultMeta = defaultMeta;
    }

    @Override
    public void send(Event event, M m) {
        // XXX：不要在send的时候修改event的值，有副作用。
        // 例如：当将同一个event发送到不同的topic的时候，buildSchemaHash会用不同的topic赋值两次，导致一些异常case
        Event copiedEvent = Event.builder().build();
        BeanUtils.copyProperties(event, copiedEvent);
        afterCommitExecutor.execute(() -> taskExecutor.execute(() -> {
            try {
                log.info("====MQ PRODUCER ====, meta={}, message = {}", m, copiedEvent.toJsonString());
                getSender().accept(copiedEvent, m);
            } catch (Exception e) {
                log.error("====MQ PRODUCER ====, meta={}, message = {}", m, copiedEvent.toJsonString(), e);
            }
        }));
    }

    @Override
    public void send(Event event) {
        send(event, defaultMeta);
    }

    /**
     * stolen from http://azagorneanu.blogspot.jp/2013/06/transaction-synchronization-callbacks.html
     * 保证在交易结束后被调用.
     */
    private static class AfterCommitExecutorImpl extends TransactionSynchronizationAdapter {
        private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();

        public void execute(Runnable runnable) {
            if (log.isInfoEnabled()) {
                log.info("Submitting new runnable {} to run after commit", runnable);
            }
            if (!TransactionSynchronizationManager.isSynchronizationActive()) {
                if (log.isInfoEnabled()) {
                    log.info("Transaction synchronization is NOT ACTIVE. Executing right now runnable {}", runnable);
                }
                runnable.run();
                return;
            }
            List<Runnable> threadRunnables = RUNNABLES.get();
            if (threadRunnables == null) {
                threadRunnables = new ArrayList<>();
                RUNNABLES.set(threadRunnables);
                TransactionSynchronizationManager.registerSynchronization(this);
            }
            threadRunnables.add(runnable);
        }

        @Override
        public void afterCommit() {
            List<Runnable> threadRunnables = RUNNABLES.get();
            if (log.isInfoEnabled()) {
                log.info("Transaction successfully committed, executing {} runnables", threadRunnables.size());
            }
            for (int i = 0; i < threadRunnables.size(); i++) {
                Runnable runnable = threadRunnables.get(i);
                if (log.isInfoEnabled()) {
                    log.info("Executing runnable {}", runnable);
                }
                try {
                    runnable.run();
                } catch (RuntimeException e) {
                    log.error("Failed to execute runnable " + runnable, e);
                }
            }
        }

        @Override
        public void afterCompletion(int status) {
            if (log.isInfoEnabled()) {
                log.info("Transaction completed with status {}", status == TransactionSynchronization.STATUS_COMMITTED ? "COMMITTED" : "ROLLED_BACK");
            }
            RUNNABLES.remove();
        }
    }
}
