package com.yeejoin.equip.mqtt;

import com.yeejoin.equip.config.MqttPropertyConfig;
import com.yeejoin.equip.entity.IndicatorData;
import com.yeejoin.equip.mqtt.message.MqttTopicEnum;
import com.yeejoin.equip.utils.ExecutorFactory;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import javax.annotation.Resource;
import java.util.Objects;
import static com.yeejoin.equip.mqtt.message.MqttConstant.*;

/**
 * 消息处理器
 *
 * @author LiuLin
 * @date 2023年08月18日 10:56
 */
@Configuration
public class MessageIntegration {

    @Resource
    private MqttPropertyConfig mqttPropertyConfig;
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{mqttPropertyConfig.getBroker()});
        options.setUserName(mqttPropertyConfig.getClientUserName());
        options.setPassword(mqttPropertyConfig.getClientPassword().toCharArray());
        options.setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT);
        // 设置心跳：1.5*20秒
        options.setKeepAliveInterval(mqttPropertyConfig.getKeepAliveInterval());
        // 设置最大并发数
        options.setMaxInflight(mqttPropertyConfig.getMaxInflight());
         options.setAutomaticReconnect(true);
        //options.setCleanSession(false);
        return options;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

    @Bean
    public MessageProducerSupport bizInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttPropertyConfig.getBizClientId(),
                mqttClientFactory(),
                mqttPropertyConfig.getBizTopic()
        );
        adapter.setCompletionTimeout(DEFAULT_COMPLETION_TIMEOUT);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(QOS_DEFAULT);
        return adapter;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                mqttPropertyConfig.getClientId(),
                mqttClientFactory()
        );
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(QOS_DEFAULT);
        return messageHandler;
    }

    @Bean
    public IntegrationFlow bizMsgFlow() {
        return IntegrationFlows
                .from(bizInbound())
                .channel(channels -> channels.executor(ExecutorFactory.buildBizExecutor()))
                .handle(MessageTransfer::mqttMessage2RawMessage)
                //根据Topic后缀进行分流
                .<IndicatorData, MqttTopicEnum>route(IndicatorData::getMqttTopicEnum,
                        mapping -> mapping
                                .subFlowMapping(MqttTopicEnum.perspective, flow -> flow
                                        .handle("handleMessageService", "processMessage")
                                        .filter(Objects::nonNull)
                                        .handle(mqttOutbound()))
                                .defaultOutputToParentFlow())
                .get();
    }
}
