Springboot Integration with MQTT

Springboot IntegrationMQTT

1. First add the mqtt configuration in the

2. Pour the dependencies in the file

		<dependency>
			<groupId></groupId>
			<artifactId>spring-integration-stream</artifactId>
		</dependency>
		<dependency>
			<groupId></groupId>
			<artifactId>spring-integration-mqtt</artifactId>
			<version>5.0.</version>
		</dependency>

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9

3.Write mqtt push message class

@Configuration
@IntegrationComponentScan
public class MqttSenderConfig {

    @Value("${}")
    private String username;

    @Value("${}")
    private String password;

    @Value("${}")
    private String hostUrl;

    @Value("${}")
    private String clientId;

    @Value("${}")
    private String defaultTopic;

    @Value("${}")
    private int completionTimeout ; //Connection timeout

    @Bean
    public MqttConnectOptions getMqttConnectOptions(){
        MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
        (true);
        (10);
        (90);
        (true);
        (username);
        (());
        (new String[]{hostUrl});
        (2);
        return mqttConnectOptions;
    }
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        (getMqttConnectOptions());
        return factory;
    }
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
        (true);
        (defaultTopic);
        return messageHandler;
    }
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54

4. Write the mqtt subscription class

@Configuration
@IntegrationComponentScan
@Slf4j
public class MqttReceiveConfig {

    @Value("${}")
    private String username;

    @Value("${}")
    private String password;

    @Value("${}")
    private String hostUrl;

    @Value("${}")
    private String clientId;

    @Value("${}")
    private String defaultTopic;

    @Value("${}")
    private int completionTimeout ; //Connection timeout

    @Autowired
    private MqttGateway mqttGateway;

    @Autowired
    private BTempChargesService bTempChargesService;

    @Bean
    public MqttConnectOptions getMqttConnectOptions(){
        MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
        (true);
        (10);
        (90);
        (true);
        (username);
        (());
        (new String[]{hostUrl});
        (2);
        return mqttConnectOptions;
    }
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        (getMqttConnectOptions());
        return factory;
    }

    //receiver channel
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    //configureclient,monitoringtopic
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId+"_inbound", mqttClientFactory(),
                        defaultTopic);
        (completionTimeout);
        (new DefaultPahoMessageConverter());
        (1);
        (mqttInputChannel());
        return adapter;
    }

    //Acquisition of data through channels
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                ("thematic：{}，Message received data：{}", ().get("mqtt_receivedTopic"), ());
            }
        };
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79

5. Test, here usedartifact


publish will be able to see the subscription replies (I've omitted some of the code here).