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";
    }
}