blame | history | patch | commit | commitdiff | ignore whitespace
rain
2024-06-14 8d9a2d656e4ae007590c622e5f7c228adacdca49 
 src/main/java/com/dji/sample/component/mqtt/handler/EventsRouter.java


@@ -1,17 +1,20 @@


package com.dji.sample.component.mqtt.handler;




import com.dji.sample.component.mqtt.model.ChannelName;


import com.dji.sample.component.mqtt.model.CommonTopicReceiver;


import com.dji.sample.component.mqtt.model.EventsMethodEnum;


import com.dji.sample.component.mqtt.model.*;


import com.dji.sample.component.mqtt.service.IMessageSenderService;


import com.fasterxml.jackson.databind.ObjectMapper;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.context.annotation.Bean;


import org.springframework.context.annotation.Configuration;


import org.springframework.integration.annotation.ServiceActivator;


import org.springframework.integration.dsl.IntegrationFlow;


import org.springframework.integration.dsl.IntegrationFlows;


import org.springframework.integration.mqtt.support.MqttHeaders;


import org.springframework.messaging.MessageHeaders;




import java.io.IOException;


import java.util.Arrays;


import java.util.Optional;




/**


 * @author sean


@@ -23,6 +26,9 @@




    @Autowired


    private ObjectMapper mapper;




    @Autowired


    private IMessageSenderService messageSenderService;




    @Bean


    public IntegrationFlow eventsMethodRouterFlow() {


@@ -42,4 +48,21 @@


                                methodEnum -> mapping.channelMapping(methodEnum, methodEnum.getChannelName())))


                .get();


    }




    @ServiceActivator(inputChannel = ChannelName.OUTBOUND_EVENTS, outputChannel = ChannelName.OUTBOUND)


    public void replyEventsOutbound(CommonTopicReceiver receiver, MessageHeaders headers) {


        if (Optional.ofNullable(receiver).map(CommonTopicReceiver::getNeedReply).flatMap(val -> Optional.of(1 != val)).orElse(true)) {


            return;


        }


        messageSenderService.publish(headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF,


                CommonTopicResponse.builder()


                        .tid(receiver.getTid())


                        .bid(receiver.getBid())


                        .method(receiver.getMethod())


                        .timestamp(System.currentTimeMillis())


                        .data(RequestsReply.success())


                        .build());




    }




}