package com.gxx.record.web.kafka;

import java.util.HashMap;
import java.util.Map;

import javax.servlet.http.HttpServletRequest;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

/**
 * kafka控制器
 * @author Gxx
 */
@Controller
@RequestMapping("/kafka/")
public class KafkaController {
	/**
	 * 日志处理器
	 */
	private final Logger logger = Logger.getLogger(KafkaController.class);

	/**
	 * kafka服务方
	 */
	@Autowired
	private KafkaProvider kafkaProvider;

	/**
	 * 发送
	 * 
	 * @param request
	 * @return
	 */
	@RequestMapping(value = "/send", produces = "application/json")
	public @ResponseBody Map<String, Object> get(HttpServletRequest request) {
		logger.info("mq发送开始");

		/**
		 * 主要参数
		 */
		String topic = request.getParameter("topic");//主题
		String value = request.getParameter("value");//对象
		boolean usePartition = false;//是否使用分区
		Integer partitionNum = 0;//总分区数
		String role = "test";//角色，用来生成key=role-value.hashCode()
		
		/**
		 * 发送mq
		 */
		Map<String, Object> res = kafkaProvider.sendMq(topic, value, usePartition, partitionNum, role);

		/**
		 * 响应
		 */
		String message = (String) res.get("message");
		String code = (String) res.get("code");

		/**
		 * 组织结果
		 */
		Map<String, Object> result = new HashMap<String, Object>();
		result.put("result", true);
		result.put("msg", code + "-" + message);
		logger.info("mq发送结束");
		return result;
	}
}