package cn.iocoder.yudao.module.iot.service.rule.data.action;
|
|
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
|
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
|
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRabbitMQConfig;
|
import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
|
import com.rabbitmq.client.Channel;
|
import com.rabbitmq.client.Connection;
|
import com.rabbitmq.client.ConnectionFactory;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
import org.springframework.stereotype.Component;
|
|
/**
|
* RabbitMQ 的 {@link IotDataRuleAction} 实现类
|
*
|
* @author HUIHUI
|
*/
|
@ConditionalOnClass(name = "com.rabbitmq.client.Channel")
|
@Component
|
@Slf4j
|
public class IotRabbitMQDataRuleAction
|
extends IotDataRuleCacheableAction<IotDataSinkRabbitMQConfig, Channel> {
|
|
@Override
|
public Integer getType() {
|
return IotDataSinkTypeEnum.RABBITMQ.getType();
|
}
|
|
@Override
|
public void execute(IotDeviceMessage message, IotDataSinkRabbitMQConfig config) throws Exception {
|
try {
|
// 1.1 获取或创建 Channel
|
Channel channel = getProducer(config);
|
// 1.2 声明交换机、队列和绑定关系
|
channel.exchangeDeclare(config.getExchange(), "direct", true);
|
channel.queueDeclare(config.getQueue(), true, false, false, null);
|
channel.queueBind(config.getQueue(), config.getExchange(), config.getRoutingKey());
|
|
// 2. 发送消息
|
channel.basicPublish(config.getExchange(), config.getRoutingKey(), null,
|
JsonUtils.toJsonByte(message));
|
log.info("[execute][message({}) config({}) 发送成功]", message, config);
|
} catch (Exception e) {
|
log.error("[execute][message({}) config({}) 发送失败]", message, config, e);
|
throw e;
|
}
|
}
|
|
@Override
|
@SuppressWarnings("resource")
|
protected Channel initProducer(IotDataSinkRabbitMQConfig config) throws Exception {
|
// 1. 创建连接工厂
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(config.getHost());
|
factory.setPort(config.getPort());
|
factory.setVirtualHost(config.getVirtualHost());
|
factory.setUsername(config.getUsername());
|
factory.setPassword(config.getPassword());
|
// 2. 创建连接
|
Connection connection = factory.newConnection();
|
// 3. 创建信道
|
return connection.createChannel();
|
}
|
|
@Override
|
protected void closeProducer(Channel channel) throws Exception {
|
if (channel.isOpen()) {
|
channel.close();
|
}
|
Connection connection = channel.getConnection();
|
if (connection.isOpen()) {
|
connection.close();
|
}
|
}
|
|
}
|
