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 AbstractWorkflowListener<H extends Ordered, F extends BasicMessageQueueFilter<Target>, Target> implements WorkflowListener {

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

    protected final List<H> businessListeners;
    protected final List<F> businessFilters;
    private final ListenerExecutor listenerExecutor;


    public AbstractWorkflowListener(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) {
        if (CollectionUtils.isEmpty(businessListeners)) {
            return;
        }

        Target convert = convert(event);

        for (F filter : businessFilters) {
            if (filter.doFilter(event, convert)) {
                return;
            }
        }

        onEvent(convert, context);
    }

    protected abstract Target convert(Event event);

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

    protected ListenerExecutor getListenerExecutor() {
        return listenerExecutor;
    }

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

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