szwlServer/src/main/java/com/szwl/controller/RabbitMqController.java

package com.szwl.controller;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.szwl.model.bo.JsonMessage;
import com.szwl.model.entity.TEquipment;
import com.szwl.service.TAdminService;
import com.szwl.service.TEquipmentService;
import io.swagger.annotations.Api;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Api(value = "/api/app_mq/rabbitMq", tags = {"MQ"})
@RestController
@RequestMapping("/api/app_mq/rabbitMq")
//@RequestMapping("/rabbitMqController")
public class RabbitMqController {
    @Autowired
    TEquipmentService tEquipmentService;
    @Autowired
    TAdminService tAdminService;
//    @Autowired
//    private RabbitAdmin rabbitAdmin;
    @Autowired
    private AmqpAdmin amqpAdmin;
//    @Autowired
//    private RabbitTemplate rabbitTemplate;
    /**
     *添加交换机
     * @param
     * @return
     */
    @GetMapping(value = "/addExchange.htm")
    @ResponseBody
    public String addExchange(String exchange,String type,Boolean durable){
        if(StringUtils.isEmpty(type)){
            return "type is null";
        }
        if(type.equals("Direct")){
            amqpAdmin.declareExchange(new DirectExchange(exchange, durable, false));

        }
        if(type.equals("Topic")){
            amqpAdmin.declareExchange(new TopicExchange(exchange, durable, false));
        }
        if(type.equals("Fanout")){
            amqpAdmin.declareExchange(new FanoutExchange(exchange, durable, false));
        }
        return "success";
//        return JsonMessage.success("成功");
    }
    /**
     *添加队列并绑定
     * @param
     * @return
     */
    @GetMapping(value = "/addQueue.htm")
    @ResponseBody
    public String addQueue(String exchange,String queue,String routingKey,Boolean durable){
        Map<String,Object> arg = new HashMap<>();
        arg.put("x-message-ttl",1800000);
        String queue1 = amqpAdmin.declareQueue(new Queue(queue, durable, false, false, arg));
        HashMap<String, Object> objectObjectHashMap = new HashMap<>();
        amqpAdmin.declareBinding(new Binding(queue,
                Binding.DestinationType.QUEUE,
                exchange, routingKey, objectObjectHashMap));
        LambdaQueryWrapper<TEquipment> query = Wrappers.lambdaQuery();
        query.eq(TEquipment::getClientId,routingKey);
        List<TEquipment> list = tEquipmentService.list(query);
        if(list.size()>0){
            TEquipment equipment = list.get(0);
            equipment.setEquimentType(exchange);
            tEquipmentService.updateById(equipment);
        }

//        return JsonMessage.success("成功");
        return "success";
    }
    /**
     *修改信道
     * @param
     * @return
     */
    @GetMapping(value = "/updateChannel.htm")
    @ResponseBody
    public String updateChannel(String clientId,String type){
        LambdaQueryWrapper<TEquipment> query = Wrappers.lambdaQuery();
        query.eq(TEquipment::getClientId,clientId);
        List<TEquipment> list = tEquipmentService.list(query);
        if(list.size()>0){
            TEquipment equipment = list.get(0);
            equipment.setChannel(type);
            tEquipmentService.updateById(equipment);
        }
        return "success";
    }

    /**
     * 修改机型重新创建MQ队列
     * @param
     * @return
     */
    @GetMapping(value = "/updateEquipmentType")
    @ResponseBody
    public String updateEquipmentType(String clientId, String equipmentType){
        if (StringUtils.isNotEmpty(clientId) && StringUtils.isNotEmpty(equipmentType)) {
            // 删除MQ队列
            amqpAdmin.deleteQueue(clientId);
            // 重新创建队列
            Map<String,Object> arg = new HashMap<>();
            arg.put("x-message-ttl",1800000);
            amqpAdmin.declareQueue(new Queue(clientId, true, false, false, arg));
            HashMap<String, Object> objectObjectHashMap = new HashMap<>();
            amqpAdmin.declareBinding(new Binding(clientId,
                    Binding.DestinationType.QUEUE,
                    equipmentType, clientId, objectObjectHashMap));
            LambdaQueryWrapper<TEquipment> query = Wrappers.lambdaQuery();
            query.eq(TEquipment::getClientId, clientId);
            TEquipment equipment = tEquipmentService.getOne(query);
            if(equipment != null){
                equipment.setEquimentType(equipmentType);
                tEquipmentService.updateById(equipment);
                return "success";
            }
            return "找不到设备";
        }

        return "参数为空";
    }
}