package cn.axzo.workflow.starter.mq.broadcast.consumer;

import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.workflow.starter.handler.execute.ListenerExecutor;
import cn.axzo.workflow.starter.mq.broadcast.filter.BasicMessageQueueFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.core.Ordered;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

public abstract class AbstractInnerWorkflowListener<H extends Ordered, F extends BasicMessageQueueFilter<Target>, Target> implements InnerWorkflowListener {

    private static final Logger log = LoggerFactory.getLogger(AbstractInnerWorkflowListener.class);

    /**
     * 接入方对 cn/axzo/workflow/starter/handler/ 包下的 EventHandler 实现
     */
    protected final List<H> businessListeners;
    /**
     * 接入方对 cn/axzo/workflow/starter/handler/filter/ 包下的EventFilter 实现
     */
    protected final List<F> businessFilters;
    protected final ListenerExecutor listenerExecutor;


    public AbstractInnerWorkflowListener(ListenerExecutor listenerExecutor,
                                         ObjectProvider<List<H>> handlerProvider,
                                         ObjectProvider<List<F>> filterProvider) {
        this.listenerExecutor = listenerExecutor;
        this.businessListeners = getOrderedBusinessListeners(handlerProvider);
        this.businessFilters = getOrderedBusinessFilters(filterProvider);
    }

    private List<F> getOrderedBusinessFilters(ObjectProvider<List<F>> filterProvider) {
        return filterProvider.getIfAvailable(Collections::emptyList)
                .stream()
                .sorted(Comparator.comparingInt(Ordered::getOrder))
                .collect(Collectors.toList());
    }

    protected List<H> getOrderedBusinessListeners(ObjectProvider<List<H>> handlerProvider) {
        return handlerProvider.getIfAvailable(Collections::emptyList)
                .stream()
                .sorted(Comparator.comparingInt(Ordered::getOrder))
                .collect(Collectors.toList());
    }

    @Override
    public void handleEvent(Event event, EventConsumer.Context context) {
        Target convert = convert(event);

        if (!CollectionUtils.isEmpty(businessListeners)) {
            for (F filter : businessFilters) {
                if (filter.doFilter(event, context, convert)) {
                    if (log.isDebugEnabled()) {
                        log.debug("【{}】filtered message, messageId: {}", filter.getClass().getSimpleName(), context.getMsgId());
                    }
                    return;
                }
            }
        }

        onEvent(convert, event, context);
    }

    protected abstract Target convert(Event event);

    protected abstract void onEvent(Target data, Event event, EventConsumer.Context context);

    @Override
    public boolean dispatch(Event event) {
        return getSupportEventCodes().contains(event.getEventCode());
    }

    protected abstract List<Event.EventCode> getSupportEventCodes();
}
