package com.yihu.hos.arbiter.services;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.WriteResult;
import com.yihu.hos.arbiter.models.BrokerServer;
import com.yihu.hos.core.http.HTTPResponse;
import com.yihu.hos.core.http.HttpClientKit;
import com.yihu.hos.web.framework.constant.ServiceFlowConstant;
import com.yihu.hos.web.framework.model.bo.ServiceFlow;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @created Airhead 2016/7/27.
 */
@Service("brokerServerService")
public class BrokerServerService {
    private static final Logger logger = LogManager.getLogger(BrokerServerService.class);

    @Autowired
    private MongoOperations mongoOperations;

    @Autowired
    private CamelContext camelContext;

    @Autowired
    private ProducerTemplate producerTemplate;
    @Autowired
    private ObjectMapper objectMapper;

    public void save(BrokerServer brokerServer) {
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));

        Update update = new Update();
        update.set("tenant", brokerServer.getTenant());
        update.set("hostName", brokerServer.getHostName());
        update.set("hostAddress", brokerServer.getHostAddress());
        update.set("port", brokerServer.getPort());
        update.set("updateTime", brokerServer.getUpdateTime());
        update.set("enable", brokerServer.isEnable());
        if (brokerServer.getServiceFlows()!=null) {
            for (BrokerServer.ServiceFlow serviceFlow : brokerServer.getServiceFlows()) {
                update.addToSet("serviceFlows", serviceFlow);
            }
        }
        WriteResult writeResult = mongoOperations.upsert(query, update, BrokerServer.class);
        if (writeResult.isUpdateOfExisting()) {
            //避免Broker重启的情况
            HTTPResponse response = HttpClientKit.post(brokerServer.getURL() + "/esb/heartbeat");
            if (response.getStatusCode() == 200 && brokerServer.isRegistered()) {
                return;
            }

            //确保Broker已经启动了流程
            brokerServer = mongoOperations.findOne(query, BrokerServer.class);
            if (brokerServer.getServiceFlows() != null) {
                return;
            }
        }

        //没有启动流程
        try {
            String msg = objectMapper.writeValueAsString(brokerServer);
            ProducerTemplate producerTemplate = createProducerTemplate();
            Map<String, Object> header = new HashMap<>();
            header.put("event", ServiceFlowConstant.BROKER_SERVER_ON);
            producerTemplate.sendBodyAndHeaders(ServiceFlowConstant.FLOW_EVENT_ENDPOINT, msg, header);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }

    /**
     * 暂不提供动态均衡算法，只是随机返回一个。
     *
     * @return
     */
    public BrokerServer get() {
        Query query = new Query();
        query.addCriteria(Criteria.where("enable").is(true));
        query.with(new Sort(new Sort.Order(Sort.Direction.DESC, "updateTime")));
        return mongoOperations.findOne(query, BrokerServer.class);
    }

    public void delete(BrokerServer brokerServer) {
        mongoOperations.remove(brokerServer);
    }

    public List<BrokerServer> get(boolean one) {
        if (one) {
            BrokerServer brokerServer = get();
            List<BrokerServer> brokerServers = new ArrayList<>();
            brokerServers.add(brokerServer);
            return brokerServers;
        }

        Query query = new Query();
        query.addCriteria(Criteria.where("enable").is(true));
        query.with(new Sort(new Sort.Order(Sort.Direction.DESC, "updateTime")));
        return mongoOperations.find(query, BrokerServer.class);
    }


    public List<BrokerServer> getBrokerList(String routeCode) {
        Query query = new Query();
        query.addCriteria(Criteria.where("serviceFlows.routeCode").is(routeCode));
        return mongoOperations.find(query, BrokerServer.class);
    }

    public void flowStarted(String msg) {

    }

    public void addServiceFlow(BrokerServer brokerServer, ServiceFlow serviceFlow) {
        BrokerServer.ServiceFlow flow = new BrokerServer.ServiceFlow();
        flow.setFlowId(serviceFlow.getId());
        flow.setRouteCode(serviceFlow.getRouteCode());
        flow.setType(serviceFlow.getFlowType());
        brokerServer.addServiceFlow(flow);
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));

        Update update = new Update();
        update.set("hostName", brokerServer.getHostName());
        update.set("hostAddress", brokerServer.getHostAddress());
        update.set("port", brokerServer.getPort());
        update.set("updateTime", brokerServer.getUpdateTime());
        update.set("enable", brokerServer.isEnable());
        update.addToSet("serviceFlows", serviceFlow);

        mongoOperations.upsert(query, update, BrokerServer.class);
    }

    public void removeServiceFlow(BrokerServer brokerServer, ServiceFlow serviceFlow) {
        Query query = new Query();
        query.addCriteria(Criteria.where("hostName").is(brokerServer.getHostName()));
        query.addCriteria(Criteria.where("hostAddress").is(brokerServer.getHostAddress()));
        query.addCriteria(Criteria.where("port").is(brokerServer.getPort()));

        Update update = new Update();
        update.set("hostName", brokerServer.getHostName());
        update.set("hostAddress", brokerServer.getHostAddress());
        update.set("port", brokerServer.getPort());
        update.set("updateTime", brokerServer.getUpdateTime());
        update.set("enable", brokerServer.isEnable());
        update.pull("serviceFlows", serviceFlow);

        mongoOperations.upsert(query, update, BrokerServer.class);
    }

    private ProducerTemplate createProducerTemplate() {
        if (producerTemplate == null) {
            producerTemplate = camelContext.createProducerTemplate();
        }

        return producerTemplate;
    }
}
