package com.yihu.hos.arbiter.services;

import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import com.yihu.hos.web.framework.constant.ServiceFlowConstant;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.zbus.broker.ZbusBroker;
import org.zbus.mq.Consumer;
import org.zbus.net.http.Message;

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

/**
 * @author Airhead
 * @since 2016/12/22.
 */
@Service
public class ProxyService {
    private static final Logger logger = LoggerFactory.getLogger(ProxyService.class);
    @Autowired
    private CamelContext camelContext;

    private ZbusBroker zbusBroker;
    private Consumer consumer;


    public void proxy(Message message, Consumer consumer) {
        Map<String, Object> header = new HashMap<>();
        header.put("event", message.getHead("event"));

        ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
        producerTemplate.sendBodyAndHeaders(ServiceFlowConstant.FLOW_EVENT_ENDPOINT, message.getBodyString(), header);
    }

    public void start() {
        if (zbusBroker == null) {
            logger.error("zbusBroker is null");
            return;
        }

        consumer = new Consumer(zbusBroker, ServiceFlowConstant.SSH);
        try {
            consumer.start(this::proxy);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }


    @Autowired
    public void setZbusBroker(ZbusBroker zbusBroker) {
        this.zbusBroker = zbusBroker;
    }

    @Override
    protected void finalize() throws Throwable {
        consumer.close();
        super.finalize();
    }
}
