package cn.axzo.msg.center.mq;

import cn.axzo.framework.rocketmq.BaseListener;
import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.framework.rocketmq.EventHandler;
import cn.axzo.msg.center.service.enums.MqMessageType;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @author yanglin
 */
public abstract class IsolationMQListener extends BaseListener
        implements RocketMQListener<MessageExt>, EventHandler, InitializingBean {

    @Autowired
    private EventConsumer eventConsumer;
    private final ConsumerIsolation isolation;
    private final Event.EventCode eventCode;

    protected IsolationMQListener(ConsumerIsolation isolation, MqMessageType mqMessageType) {
        this(isolation, mqMessageType.getEventCode());
    }

    protected IsolationMQListener(ConsumerIsolation isolation, Event.EventCode eventCode) {
        this.isolation = isolation;
        this.eventCode = eventCode;
    }

    @Override
    public void onMessage(MessageExt message) {
        ConsumerIsolation.setIsolation(isolation);
        try {
            super.onEvent(message, eventConsumer);
        } finally {
            ConsumerIsolation.clearIsolation();
        }
    }

    @Override
    public final void onEvent(Event event, EventConsumer.Context context) {
        if (ConsumerIsolation.getIsolation() == isolation)
            onEventImpl(event, context);
    }

    @Override
    public void afterPropertiesSet() {
        eventConsumer.registerHandler(eventCode, this);
    }

    public abstract void onEventImpl(Event event, EventConsumer.Context context);
}