Merge branch 'master' of luofaqiang/esb into master

src/main/java/com/yihu/hos/services/ServiceFlowEventService.java

@@ -1,5 +1,6 @@
package com.yihu.hos.services;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.common.constants.ContextAttributes;
import com.yihu.hos.core.log.Logger;
@@ -10,12 +11,10 @@ import com.yihu.hos.web.framework.model.bo.ServiceFlow;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Component;
import org.zbus.broker.ZbusBroker;
import org.zbus.mq.Producer;
import org.zbus.net.http.Message;
import javax.jms.Queue;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
/**
 * @created Airhead 2016/8/2.
@@ -32,13 +31,6 @@ public class ServiceFlowEventService {
    @Autowired
    private ObjectMapper objectMapper;
    private ZbusBroker zbusBroker;
    @Autowired
    public void setZbusBroker(ZbusBroker zbusBroker) {
        this.zbusBroker = zbusBroker;
    }
    /**
     * admin启动时，触发一次所有流程更新事件，用于重启整个服务的情况。
     * 同时解决Broker中启动多个采集任务的问题。
@@ -80,26 +72,17 @@ public class ServiceFlowEventService {
    private void sendMsg(String event, ServiceFlow serviceFlow) {
        if (zbusBroker == null) {
            logger.error("zbusBroker is null.");
            return;
        }
        try {
            String msg = objectMapper.writeValueAsString(serviceFlow);
            String tenant = LocalContext.getContext().getAttachment(ContextAttributes.TENANT_NAME);
            Producer producer = new Producer(zbusBroker, ServiceFlowConstant.SSH + "@" + tenant);
            producer.createMQ();    //确定为创建消息队列需要显示调用
            Message message = new Message();
            message.setHead("event", event);
            message.setHead("tenant", tenant);
            message.setMethod("POST");
            message.setBody(msg);
            producer.sendSync(message);
        } catch (IOException | InterruptedException e) {
            logger.error(e.getMessage());
            Map<String, Object> header = new HashMap<>();
            String attachment = LocalContext.getContext().getAttachment(ContextAttributes.TENANT_NAME);
            serviceFlow.setTenant(attachment);
            header.put("tenant", attachment);
            header.put("event", event);
                    this.jmsMessagingTemplate.convertAndSend(this.queue, msg, header);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
}

src/main/java/com/yihu/hos/services/ServiceShellEventService.java

@@ -1,5 +1,6 @@
package com.yihu.hos.services;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yihu.hos.common.constants.ContextAttributes;
import com.yihu.hos.core.log.Logger;
@@ -10,17 +11,14 @@ import com.yihu.hos.web.framework.model.bo.ServiceShell;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Component;
import org.zbus.broker.ZbusBroker;
import org.zbus.mq.Producer;
import org.zbus.net.http.Message;
import javax.annotation.Resource;
import javax.jms.Queue;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
/**
 * 发送shell命令消息到ＭＱ
 *
 *  发送shell命令消息到ＭＱ
 * @author HZY
 * @vsrsion 1.0
 * Created at 2017/1/5.
@@ -37,13 +35,6 @@ public class ServiceShellEventService {
    @Autowired
    private ObjectMapper objectMapper;
    private ZbusBroker zbusBroker;
    @Autowired
    public void setZbusBroker(ZbusBroker zbusBroker) {
        this.zbusBroker = zbusBroker;
    }
    public void serviceShellSend(ServiceShell serviceShell) {
        serviceShell.setType(ServiceFlowConstant.ARBITER_SHELL_SEND);
        this.sendMsg(ServiceFlowConstant.ARBITER_SHELL_SEND, serviceShell);
@@ -56,28 +47,20 @@ public class ServiceShellEventService {
    }
    private void sendMsg(String event, ServiceShell serviceShell) {
        if (zbusBroker == null) {
            logger.error("zbusBroker is null.");
            return;
        }
        try {
            Map<String, Object> header = new HashMap<>();
            String attachment = LocalContext.getContext().getAttachment(ContextAttributes.TENANT_NAME);
            serviceShell.setTenant(attachment);
            String msg = objectMapper.writeValueAsString(serviceShell);
            String tenant = LocalContext.getContext().getAttachment(ContextAttributes.TENANT_NAME);
            Producer producer = new Producer(zbusBroker, ServiceFlowConstant.SHELL_REQUEST + "@" + tenant);
            producer.createMQ();    //确定为创建消息队列需要显示调用
            Message message = new Message();
            message.setHead("event", event);
            message.setHead("tenant", tenant);
            message.setMethod("POST");
            message.setBody(msg);
            producer.sendSync(message);
        } catch (IOException | InterruptedException e) {
            logger.error(e.getMessage());
            header.put("tenant", attachment);
            header.put("event", event);
            this.jmsMessagingTemplate.convertAndSend(this.shellQueue, msg, header);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
}