增加订购topic和发送消息的restful接口

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/action/ConsumerAction.java

@@ -0,0 +1,48 @@
package com.ai.ipu.ddmp.data.action;
import javax.servlet.http.HttpServletRequest;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.ai.aif.msgframe.consumer.MfConsumerClient;
import com.ai.ipu.basic.string.StringUtil;
import com.ai.ipu.ddmp.data.config.DataConfig;
import com.ai.ipu.ddmp.data.util.AppConstant;
import com.ai.ipu.ddmp.data.util.ReturnCode;
import com.ai.ipu.ddmp.data.util.SpringUtil;
import com.alibaba.fastjson.JSONObject;
@RestController
@RequestMapping("/rest/Consumer")
public class ConsumerAction {
	/*
	 * 必须先订购，再发送消息，否则消息无法被接收到
	 */
	@RequestMapping(value = "/Subscribe")
    public String subscribeTopic(HttpServletRequest req) {
		//获取参数
        String topic = req.getParameter("topic"), tag = req.getParameter("tag");
		// 返回处理结果
        JSONObject obj = new JSONObject();
        obj.put(AppConstant.result.RETURN_CODE, ReturnCode.SUCCESS.getCode());
        obj.put(AppConstant.result.RETURN_MSG, ReturnCode.SUCCESS.getDescription());
        if (!StringUtil.isEmpty(topic) 
        		&&(((DataConfig) SpringUtil.getBean("dataConfig")).getOrigTopics().indexOf(topic)>=0))
        	if (StringUtil.isEmpty(tag))
        		MfConsumerClient.subscribe(topic, "*","com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl");
        	else
        		MfConsumerClient.subscribe(topic, tag,"com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl");
        else
        {
        	obj.put(AppConstant.result.RETURN_CODE, ReturnCode.PARAM_ERROR.getCode());
        	if (StringUtil.isEmpty(topic)) 
        		obj.put(AppConstant.result.RETURN_MSG, ReturnCode.PARAM_ERROR.getDescription()+",订购失败!topic为空");
        	else
        		obj.put(AppConstant.result.RETURN_MSG, ReturnCode.PARAM_ERROR.getDescription()+",订购失败!topic"+topic+"不在订购列表中");
        }
        return obj.toString();
	}
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/action/ProducerAction.java

@@ -0,0 +1,68 @@
package com.ai.ipu.ddmp.data.action;
import java.rmi.RemoteException;
import javax.servlet.http.HttpServletRequest;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.ai.aif.msgframe.MfProducerClient;
import com.ai.aif.msgframe.common.exception.MsgFrameException;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
import com.ai.ipu.basic.string.StringUtil;
import com.ai.ipu.ddmp.data.util.AppConstant;
import com.ai.ipu.ddmp.data.util.ReturnCode;
import com.alibaba.fastjson.JSONObject;
@RestController
@RequestMapping("/rest/Producer")
public class ProducerAction {
	@RequestMapping(value = "/SendTextMessage")
    public String sendTextMessage(HttpServletRequest req) {
		//获取参数
        String topic = req.getParameter("topic"), msg = req.getParameter("msg");
        String tag = req.getParameter("tag");
		// 返回处理结果
        JSONObject obj = new JSONObject();
        obj.put(AppConstant.result.RETURN_CODE, ReturnCode.SUCCESS.getCode());
        obj.put(AppConstant.result.RETURN_MSG, ReturnCode.SUCCESS.getDescription());
        
        if (!StringUtil.isEmpty(topic))
        {
        	if (!StringUtil.isEmpty(msg))
        	{
        		MfProducerClient client = new MfProducerClient();
    			MsgFTextMessage message = new MsgFTextMessage(msg);
    			if (!StringUtil.isEmpty(tag)) 
    				message.setFilterTag(tag);// 设置消息tag
    			try {
    				client.send(topic, message);
    			} catch (RemoteException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    				obj.put(AppConstant.result.RETURN_CODE, ReturnCode.INSIDE_ERROR.getCode());
                	obj.put(AppConstant.result.RETURN_MSG, ReturnCode.INSIDE_ERROR.getDescription()+",发送文本消息失败!无法连接到MQ");
    			} catch (MsgFrameException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    				obj.put(AppConstant.result.RETURN_CODE, ReturnCode.INSIDE_ERROR.getCode());
                	obj.put(AppConstant.result.RETURN_MSG, ReturnCode.INSIDE_ERROR.getDescription()+",发送文本消息失败!");
    			}
        	}
        	else
        	{
            	obj.put(AppConstant.result.RETURN_CODE, ReturnCode.PARAM_ERROR.getCode());
            	obj.put(AppConstant.result.RETURN_MSG, ReturnCode.PARAM_ERROR.getDescription()+",发送文本消息失败!msg为空");
            }
        }
        else
        {
        	obj.put(AppConstant.result.RETURN_CODE, ReturnCode.PARAM_ERROR.getCode());
        	obj.put(AppConstant.result.RETURN_MSG, ReturnCode.PARAM_ERROR.getDescription()+",发送文本消息失败!topic为空");
        }
			
        return obj.toString();
	}
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/config/DataConfig.java

@@ -38,6 +38,9 @@ public class DataConfig {
	public String[] getTopics() {
		return topics.split(getSeperator());
	}
	public String getOrigTopics() {
		return topics;
	}
	public void setTopics(String topics) {
		this.topics = topics;
	}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/util/AppConstant.java

@@ -0,0 +1,9 @@
package com.ai.ipu.ddmp.data.util;
public class AppConstant {
	public interface result{
		public static final String RETURN_CODE = "returnCode";
		public static final String RETURN_MSG = "returnMsg";
		
	}
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/util/ReturnCode.java

@@ -0,0 +1,23 @@
package com.ai.ipu.ddmp.data.util;
public enum ReturnCode {
	SUCCESS("0000", "成功"),
    INSIDE_ERROR("0111","系统内部错误"),
    PARAM_ERROR("0112","参数有误");
	
	private String code;
	private String description;
	private ReturnCode(String code, String description) {
		this.code = code;
		this.description = description;
	}
	public String getCode() {
		return this.code;
	}
	
	public String getDescription() {
		return this.description;
	}
}