package tech.powerjob.remote.akka;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.DeadLetter;
import akka.actor.Props;
import akka.routing.RoundRobinPool;
import com.google.common.collect.Maps;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import tech.powerjob.common.serialize.JsonUtils;
import tech.powerjob.remote.framework.actor.ActorInfo;
import tech.powerjob.remote.framework.base.Address;
import tech.powerjob.remote.framework.cs.CSInitializer;
import tech.powerjob.remote.framework.cs.CSInitializerConfig;
import tech.powerjob.remote.framework.transporter.Transporter;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * AkkaCSInitializer
 *
 * @author tjq
 * @since 2022/12/31
 */
@Slf4j
public class AkkaCSInitializer implements CSInitializer {

    private ActorSystem actorSystem;
    private CSInitializerConfig config;

    @Override
    public String type() {
        return tech.powerjob.common.enums.Protocol.AKKA.name();
    }

    @Override
    public void init(CSInitializerConfig config) {

        this.config = config;

        Address bindAddress = config.getBindAddress();
        log.info("[PowerJob-AKKA] bindAddress: {}", bindAddress);

        // 初始化 ActorSystem（macOS上 new ServerSocket 检测端口占用的方法并不生效，可能是AKKA是Scala写的缘故？没办法...只能靠异常重试了）
        Map<String, Object> overrideConfig = Maps.newHashMap();

        Address externalAddress = config.getExternalAddress();

        if (externalAddress == null || StringUtils.equalsIgnoreCase(externalAddress.toFullAddress(), bindAddress.toFullAddress())) {
            overrideConfig.put("akka.remote.artery.canonical.hostname", bindAddress.getHost());
            overrideConfig.put("akka.remote.artery.canonical.port", bindAddress.getPort());
            log.info("[PowerJob-AKKA] not exist externalIp, overrideConfig: {}", overrideConfig);
        } else {
            overrideConfig.put("akka.remote.artery.canonical.hostname", externalAddress.getHost());
            overrideConfig.put("akka.remote.artery.canonical.port", externalAddress.getPort());

            overrideConfig.put("akka.remote.artery.bind.hostname", "0.0.0.0");
            overrideConfig.put("akka.remote.artery.bind.port", bindAddress.getPort());

            log.info("[PowerJob-AKKA] exist externalAddress[{}], final overrideConfig: {}", externalAddress, overrideConfig);
        }

        Config akkaBasicConfig = ConfigFactory.load(AkkaConstant.AKKA_CONFIG);
        Config akkaFinalConfig = ConfigFactory.parseMap(overrideConfig).withFallback(akkaBasicConfig);

        log.info("[PowerJob-AKKA] try to start AKKA System.");

        // 启动时绑定当前的 actorSystemName
        String actorSystemName = AkkaConstant.fetchActorSystemName(config.getServerType());
        this.actorSystem = ActorSystem.create(actorSystemName, akkaFinalConfig);

        // 处理系统中产生的异常情况
        ActorRef troubleshootingActor = actorSystem.actorOf(Props.create(AkkaTroubleshootingActor.class), "troubleshooting");
        actorSystem.eventStream().subscribe(troubleshootingActor, DeadLetter.class);

        log.info("[PowerJob-AKKA] initialize actorSystem[{}] successfully!", actorSystem.name());
    }

    @Override
    public Transporter buildTransporter() {
        return new AkkaTransporter(actorSystem);
    }

    @Override
    public void bindHandlers(List<ActorInfo> actorInfos) {
        int cores = Runtime.getRuntime().availableProcessors();
        actorInfos.forEach(actorInfo -> {
            String rootPath = actorInfo.getAnno().path();
            AkkaMappingService.ActorConfig actorConfig = AkkaMappingService.parseActorName(rootPath);

            log.info("[PowerJob-AKKA] start to process actor[path={},config={}]", rootPath, JsonUtils.toJSONString(actorConfig));

            actorSystem.actorOf(AkkaProxyActor.props(actorInfo)
                    .withDispatcher("akka.".concat(actorConfig.getDispatcherName()))
                    .withRouter(new RoundRobinPool(cores)), actorConfig.getActorName());

        });
    }

    @Override
    public void close() throws IOException {
        actorSystem.terminate();
    }
}