svr-eip-broker/src/main/java/com/yihu/hos/broker/services/camel/CamelStartBoot.java

package com.yihu.hos.broker.services.camel;

import com.yihu.hos.broker.common.log.TracerFormatter;
import com.yihu.hos.broker.services.BrokerServerService;
import com.yihu.hos.broker.util.lang.DynamicClassLoader;
import com.yihu.hos.core.log.Logger;
import com.yihu.hos.core.log.LoggerFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.interceptor.DefaultTraceFormatter;
import org.apache.camel.processor.interceptor.Tracer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.stereotype.Component;

import java.util.concurrent.SynchronousQueue;

/**
 * @author Airhead
 * @since 2016/12/9.
 */
@Component
public class CamelStartBoot {
    private static Logger logger = LoggerFactory.getLogger(CamelStartBoot.class);

    private BrokerServerService brokerServerService;
    private CamelContext context = SystemCamelContext.getContext();

    @Autowired
    private AutowireCapableBeanFactory capableBeanFactory;

    @Autowired
    public void setBrokerServerService(BrokerServerService brokerServerService) {
        this.brokerServerService = brokerServerService;
    }

    public void start() {
        logger.info("Apache Camel Context 启动...");

        try {
            context.setStreamCaching(true); //确保outBody可以消费多次。参考链接：http://camel.apache.org/why-is-my-message-body-empty.html
            context.setUseMDCLogging(true);

            Tracer tracer = new Tracer();
            tracer.setTraceOutExchanges(true);

            DefaultTraceFormatter formatter = new TracerFormatter();
            formatter.setShowHeaders(true);
            formatter.setShowBody(true);
            formatter.setShowBodyType(true);
            formatter.setShowOutHeaders(true);
            formatter.setShowOutBody(true);
            formatter.setShowOutBodyType(true);
            tracer.setFormatter(formatter);

            context.addInterceptStrategy(tracer);
            context.setTracing(true);

            context.getGlobalOptions().put(Exchange.LOG_DEBUG_BODY_STREAMS, "true");

            context.start();
            logger.info("Apache Camel Context 启动完成...");

            brokerServerService.login();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("Apache Camel Context 启动失败。");
        }

    }

    /**
     * serveProcessor 创建Processor实例
     */
    public void serveProcessor() {
        try {
            SynchronousQueue<String> processorQueue = SystemCamelContext.getProcessorQueue();
            String className;
            while ((className = processorQueue.take()) != null) {
                try {
                    DynamicClassLoader dynamicClassLoader = new DynamicClassLoader(SystemCamelContext.getResource(this).getPath());
                    Class<?> processorClass = dynamicClassLoader.loadClass(className);
                    if (processorClass != null) {
                        Object o = processorClass.newInstance();
                        capableBeanFactory.autowireBean(o);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    logger.error("serveProcessor----加载数据Class异常11。。");
                }
                logger.info(className);
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("serveProcessor----加载数据Class异常11。。");
        }
    }

    /**
     * serveRouter 创建Router实例
     */
    public void serveRouter() {
        try {
            SynchronousQueue<String> routerQueue = SystemCamelContext.getRouterQueue();
            String className;
            while ((className = routerQueue.take()) != null) {
                try {
                    DynamicClassLoader dynamicClassLoader = new DynamicClassLoader(SystemCamelContext.getResource(this).getPath());
                    Class<RouteBuilder> routeBuilderClass = (Class<RouteBuilder>) dynamicClassLoader.loadClass(className);
                    if (routeBuilderClass != null) {
                        RouteBuilder routeBuilder = routeBuilderClass.newInstance();
                        capableBeanFactory.autowireBean(routeBuilder);

                        context.addRoutes(routeBuilder);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    logger.error("serveRouter----加载数据Class异常11。");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("serveRouter----加载数据Class异常22。");
        }

    }

    public void shutdown() {
        brokerServerService.logout();
    }
}