여기서 아웃바운드 채널은 (Spring Application → 외부시스템(MQTT Broker)) 메시지 전달을 위한 채널
@Configuration
public class MqttConfig {
@Value("${spring.mqtt.broker-url}")
private String brokerUrl;
@Value("${spring.mqtt.client-id}")
private String clientId;
@Value("${spring.mqtt.topic}")
private String topic;
// 클라이언트 팩토리
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{brokerUrl});
factory.setConnectionOptions(options);
return factory;
}
// 채널 구성
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(topic);
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
// 메시지 전송 채널 지정
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MyGateway {
void sendToMqtt(String data);
}
}
MqttService
process() 메서드가 호출될 때 mqtt broker로 메시지를 전송한다.
@Service
@RequiredArgsConstructor
@Slf4j
public class MqttService {
private final MqttConfig.MyGateway myGateway;
public void process() {
String msg = "Hello MQTT!";
myGateway.sendToMqtt(msg);
}
}
ⅲ. Inbound 예제
MqttConfig
여기서 인바운드 채널은 Topic을 구독을 위한 채널 구성(MQTT Broker → Spring Application)
@Configuration
public class MqttConfig {
private static final String BROKER_URL = "tcp://localhost:1883"; // 브로커 주소
private static final String MQTT_CLIENT_ID = MqttAsyncClient.generateClientId(); // 클라이언트 고유식볋자
private static final String TOPIC_FILTER = "test/topic"; // Topic
@Bean
public MessageChannel mqttInputChannel() {
// Point to Point Channel 중 가장 기본
// messageHandler에게 Message를 전송
// DirectChannel을 구독하는 핸들러 하나에게만 브로드 캐스트
// PublishSubscribeChannel은 Publish/Subscribe Channel로
// 해당 채널을 구독한 모든 핸들러에게 브로드 캐스트
return new DirectChannel();
}
@Bean
public MessageProducer inbound() {
// 메시지 수신을 위한 채널을 구성
// 생성자를 통해 topic을 여러개 추가할 수 있음.
// addTopic() 메서드도 존재함.
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(BROKER_URL, MQTT_CLIENT_ID, TOPIC_FILTER);
adapter.setCompletionTimeout(5000); // 메시지 처리가 5초를 초과하면 타임아웃으로 간주
adapter.setConverter(new DefaultPahoMessageConverter()); // MQTT 메시지를 Spring Integration 메시지로 변환
adapter.setQos(1); // Qos 설정
adapter.setOutputChannel(mqttInputChannel()); // Spring Integration 채널 설정
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler messageHandler() {
// MQTT 클라이언트에 의해 수신된 페이로드는
// Broker -> inbound ➤ mqttInputChannel ➤ MessageHandler
// 순으로 이동되어 MessageHandle에서 메시지를 확인할 수 있음.
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
// TODO Auto-generated method stub
System.out.println(message.getHeaders());
System.out.println("Topic: " + message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
System.out.println("Payload: " + message.getPayload());
}
};
}
}
message.getHeaders 메타정보에는 아래와 같은 정보가 포함되어 있다.
mqtt_receivedRetained=false, => Retained flag : true이면 메시지가 브로커에 보관
mqtt_id=0, => 패킷 식별자
mqtt_duplicate=false, => true이면 메시지가 중복 전송된 것
id=230df40a-46b7-3695-9776-ab1f9bed887b, => 메시지 식별자
