package com.sonic.common.event;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

import com.sonic.common.exception.BizException;
import com.sonic.common.rpc.GlobalResultCode;
import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;

import lombok.extern.slf4j.Slf4j;

/**
 * 事件处理器仓库
 * 可以在多个地方使用.
 */
@Slf4j
public class EventHandlerRepository {
    final protected ListMultimap<Event.EventCode, EventHandler> handlers = ArrayListMultimap.create();
    private final BiConsumer<Exception, String> exceptionHandler;

    public EventHandlerRepository(BiConsumer<Exception, String> exceptionHandler) {
        this.exceptionHandler = exceptionHandler;
    }

    public EventHandlerRepository() {
        this(null);
    }

    public EventHandlerRepository registerHandler(Event.EventCode eventCode, EventHandler eventHandler) {
        Objects.requireNonNull(eventCode);
        Objects.requireNonNull(eventHandler);

        handlers.put(eventCode, eventHandler);
        return this;
    }

    public EventHandlerRepository registerHandlers(List<Event.EventCode> eventCodes, EventHandler eventHandler) {
        Objects.requireNonNull(eventHandler);
        eventCodes.forEach(e -> handlers.put(e, eventHandler));
        return this;
    }

    /**
     * 处理事件
     * @param event 事件
     * @return 如果有注册的handler处理事件返回true, 否则false
     */
    public boolean process(Event event, Map<String, Object> ext) {
        Stopwatch stopwatch = Stopwatch.createUnstarted();
        List<EventHandler> eventHandlers = handlers.get(event.getEventCode());
        eventHandlers.stream().forEach(handler -> {
            try {
                stopwatch.start();
                String clazzName = handler.getClass().getCanonicalName();
                log.info("====MQ CONSUMER====, start handling by {}", clazzName);
                handler.onEvent(event);

                long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
                log.info("====MQ CONSUMER====, handled by {}, cost {} millis", clazzName, elapsed);

                Number maxElapsed = (Number) ext.getOrDefault("maxElapsedMillis", 10_000);
                if (elapsed > maxElapsed.longValue()) {
                    String msg = String.format("take too long %d millis for %s to handle %s",
                            elapsed, clazzName, event.toJsonString());
                    handleException(new BizException(GlobalResultCode.EVENT_HANDLED_TIMEOUT), msg);
                }
            } catch (Exception ex) {
                log.error("====MQ CONSUMER====, handle event error, event = {}", event.toJsonString(), ex);
                handleException(ex, event.toJsonString());
            } finally {
                // stopwatch必须reset()，否则下一次stopwatch.start()会报错
                stopwatch.reset();
            }
        });
        return !eventHandlers.isEmpty();
    }

    private void handleException(Exception ex, String msg) {
        if (exceptionHandler != null) {
            exceptionHandler.accept(ex, msg);
        }
    }

    public boolean isHandled(Event event) {
        Preconditions.checkArgument(event != null);
        Preconditions.checkArgument(event.getEventCode() != null);

        return handlers.containsKey(event.getEventCode());
    }
}
