package com.usky.issue.service.config.mqtt;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;

/**
 * @author han
 * @date 2025/03/20 14:30
 */
@ConditionalOnProperty(prefix = "mqtt", value = {"enabled"}, havingValue = "true")
@Configuration
public class MqttInConfig {
    @Autowired
    private MqttBaseConfig mqttBaseConfig;

    public static final String CHANNEL_NAME_INPUT = "mqttInputChannel";

    @Bean(name = CHANNEL_NAME_INPUT)
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }


    /**
     * 消息订阅绑定-消费者
     *
     * @return
     */
    @Bean
    public MessageProducer inbound() {
        String[] tops = mqttBaseConfig.getMsgTopic().split(",");
        String clientId = "h-agbox-mqtt-in-" + System.currentTimeMillis();
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,
                mqttBaseConfig.mqttClientFactory(), tops);
        adapter.setCompletionTimeout(mqttBaseConfig.getCompletionTimeout());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
}
