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

import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.workflow.common.enums.ProcessMessagePushEventEnum;
import cn.axzo.workflow.common.model.response.mq.MessagePushDTO;
import cn.axzo.workflow.starter.handler.MessageNotificationEventHandler;
import cn.axzo.workflow.starter.handler.execute.ListenerExecutor;
import cn.axzo.workflow.starter.handler.filter.MessageNotificationEventFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @author wangli
 * @since 2024/5/21 15:53
 */
public class InnerNotificationEventListener extends AbstractWorkflowListener<MessageNotificationEventHandler, MessageNotificationEventFilter, MessagePushDTO> {
    private final static Logger log = LoggerFactory.getLogger(InnerNotificationEventListener.class);

    public final static Supplier<List<Event.EventCode>> SUPPORTED_EVENT_CODES_SUPPLIER = () ->
            Arrays.stream(ProcessMessagePushEventEnum.values()).map(ProcessMessagePushEventEnum::getEventCode)
                    .collect(Collectors.toList());

    public InnerNotificationEventListener(ListenerExecutor executor, ObjectProvider<List<MessageNotificationEventHandler>> provider, ObjectProvider<List<MessageNotificationEventFilter>> filterProvider) {
        super(executor, provider, filterProvider);
    }

    @Override
    protected MessagePushDTO convert(Event event) {
        return event.normalizedData(MessagePushDTO.class);
    }

    @Override
    public void onEvent(MessagePushDTO instanceDTO, EventConsumer.Context context) {
        ProcessMessagePushEventEnum type = instanceDTO.getType();
        for (MessageNotificationEventHandler noticeListener : businessListeners) {
            Consumer<MessagePushDTO> consumer = null;
            switch (type) {
                case PROCESS_PUSH_NOTICE:
                    consumer = noticeListener::pushNotice;
                    break;
                case PROCESS_PUSH_PENDING:
                    consumer = noticeListener::pushPending;
                    break;
                case PROCESS_PUSH_PENDING_COMPLETE:
                    consumer = noticeListener::completePending;
                    break;
                case PROCESS_PUSH_PENDING_ROLLBACK:
                    consumer = noticeListener::rollbackPending;
                    break;
                case PROCESS_CARBON_COPY:
                    consumer = noticeListener::carbonCopy;
                    break;
                case PROCESS_CARBON_COPY_COMPLETE:
                    consumer = noticeListener::carbonCopyComplete;
                    break;
                case PROCESS_PUSH_SMS:
                    consumer = noticeListener::pushSms;
                    break;
                default:
                    log.warn("unknown message event type: {}", type);
            }
            getListenerExecutor().execute(consumer, instanceDTO);
        }
    }

    @Override
    protected List<Event.EventCode> getSupportEventCodes() {
        return SUPPORTED_EVENT_CODES_SUPPLIER.get();
    }


}
