MQTTを使用してデバイスと通信し、そのデバイスに対してハートビートを送信するSpring Bootの実装


## **実装の概要**

1. **デバイスがMQTTを使ってトピック `/device/register` にメッセージを送信**
    - メッセージの内容に `deviceId` を含む。
2. **Spring Bootがデバイスを登録**
    - デバイスの `deviceId` を保持。
3. **スケジューラーでMQTT経由でハートビートを送信**
    - 一定間隔（例: 30秒ごと）で `/device/heartbeat/{deviceId}` にMQTTメッセージを送信。

---

## **1. 必要なライブラリ（`pom.xml`）**

```ruby:xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>

```

---

## **2. MQTT設定 (`MqttConfig.java`)**

```ruby:java

package com.example.mqtt;

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.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.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Configuration
public class MqttConfig {

    private static final String MQTT_BROKER_URL = "tcp://localhost:1883";
    private static final String CLIENT_ID = "spring-mqtt-client";

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{MQTT_BROKER_URL});
        options.setCleanSession(true);
        factory.setConnectionOptions(options);
        return factory;
    }

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

    @Bean
    public MqttPahoMessageDrivenChannelAdapter inboundAdapter(MqttPahoClientFactory mqttClientFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(CLIENT_ID + "_in", mqttClientFactory, "/device/register");
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler messageHandler() {
        return message -> {
            String payload = message.getPayload().toString();
            System.out.println("Received MQTT message: " + payload);
        };
    }

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

    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MqttPahoMessageHandler outboundHandler(MqttPahoClientFactory mqttClientFactory) {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(CLIENT_ID + "_out", mqttClientFactory);
        handler.setAsync(true);
        return handler;
    }
}

```

---

## **3. デバイス登録とハートビート送信 (`DeviceService.java`)**

```ruby:java

package com.example.mqtt;

import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.HashSet;
import java.util.Set;

@Service
public class DeviceService {

    private final MessageChannel mqttOutputChannel;
    private final Set<String> registeredDevices = new HashSet<>();

    public DeviceService(MessageChannel mqttOutputChannel) {
        this.mqttOutputChannel = mqttOutputChannel;
    }

    public void registerDevice(String deviceId) {
        registeredDevices.add(deviceId);
        System.out.println("Device registered: " + deviceId);
    }

    @Scheduled(fixedRate = 30000)
    public void sendHeartbeat() {
        for (String deviceId : registeredDevices) {
            String topic = "/device/heartbeat/" + deviceId;
            String payload = "heartbeat";

            Message<String> message = MessageBuilder.withPayload(payload)
                    .setHeader(MqttHeaders.TOPIC, topic)
                    .build();

            mqttOutputChannel.send(message);
            System.out.println("Sent heartbeat to: " + topic);
        }
    }
}

```

---

## **4. デバイス登録を処理する (`MqttMessageHandler.java`)**

```ruby:java
package com.example.mqtt;

import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.stereotype.Component;

@Component
public class MqttMessageHandler {

    private final DeviceService deviceService;

    public MqttMessageHandler(DeviceService deviceService) {
        this.deviceService = deviceService;
    }

    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMqttMessage(String payload) {
        System.out.println("Received device registration: " + payload);
        deviceService.registerDevice(payload);
    }
}

```

---

## **5. `@EnableScheduling` を追加 (`MqttApplication.java`)**

```ruby:java
package com.example.mqtt;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class MqttApplication {
    public static void main(String[] args) {
        SpringApplication.run(MqttApplication.class, args);
    }
}

```

---

## **動作確認**

### **1. MQTTブローカーの起動**

ローカルで `mosquitto` を動かしている場合:

```ruby:bash
mosquitto -c /etc/mosquitto/mosquitto.conf

```

Dockerで `Eclipse Mosquitto` を使う場合:

```ruby:bash
docker run -d -p 1883:1883 eclipse-mosquitto

```

---

### **2. デバイスが登録を送信**

MQTTクライアント（`mosquitto_pub`）でデバイスを登録:

```ruby:bash
mosquitto_pub -h localhost -t "/device/register" -m "device123"

```

Spring Bootのコンソール:

```ruby:yaml
Received device registration: device123
Device registered: device123

```

---

### **3. ハートビートの受信**

30秒ごとに `mosquitto_sub` で受信を確認:

```ruby:bash
mosquitto_sub -h localhost -t "/device/heartbeat/device123"

```

期待される出力:

```ruby:nginx
heartbeat

```

---

## **まとめ**

✅ **デバイスは MQTT を使って `/device/register` に登録**

✅ **Spring Bootは登録されたデバイスを保存**

✅ **30秒ごとに `/device/heartbeat/{deviceId}` にハートビートを送信**

MQTTでデバイスと通信しながら、Spring Bootで管理・自動ハートビート送信を実装しました！🚀