Spring 集成：MQTT 集成测试基础

Question

我尝试测试使用 Spring 引导和 Spring 集成创建的简单 MQTT 侦听器，但无法正常工作。我尝试了很多方法。最有希望的是：

@RunWith(SpringRunner.class)
@SpringBootTest
public class BasicMqttTest {

@Value("${mqtt.client.id}")
private String mqttClientId;

@Value("${mqtt.state.topic}")
private String mqttTopic;

@Autowired
MqttPahoClientFactory mqttPahoClientFactory;

protected IMqttClient client;

@Before
public void setUp() throws Exception {
    client = mqttPahoClientFactory.getClientInstance(mqttPahoClientFactory.getConnectionOptions().getServerURIs()[0], mqttClientId);
    client.connect();
}

@After
public void tearDown() throws Exception {
    client.disconnect();
    client.close();
}

@Test
public void contextLoads() throws Exception {
    MqttMessage mqttMessage = new MqttMessage();
    mqttMessage.setPayload("MQTT!".getBytes());
    client.publish(mqttTopic, mqttMessage);
}
}

但是，测试运行 2018-07-12 16:53:50.937 ERROR 21160 --- [T Rec: consumer] .m.i.MqttPahoMessageDrivenChannelAdapter : Lost connection: Verbindung wurde getrennt; retrying... ，我没有看到任何打印出来的东西。代码大部分来自：https://github.com/spring-projects/spring-integration-samples/tree/master/basic/mqtt .该示例运行良好，但我需要能够编写适当的集成测试。

配置为：

@Value("${mqtt.server.uri}")
private String mqttServerUri;

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

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

@Value("${mqtt.client.id}")
private String mqttClientId;

@Value("${mqtt.state.topic}")
private String mqttTopic;

@Value("${mqtt.completion.timeout}")
private Integer mqttCompletionTimeout;

@Value("${mqtt.quality.of.service}")
private Integer mqttQualityOfService;

@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setServerURIs(new String[]{mqttServerUri});
    options.setUserName(mqttUsername);
    options.setPassword(mqttPassword.toCharArray());
    factory.setConnectionOptions(options);
    return factory;
}

@Bean
@Qualifier("MqttInboundChannel")
public MessageProducerSupport mqttInbound() {
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
            mqttClientId,
            mqttClientFactory(),
            mqttTopic
    );
    adapter.setCompletionTimeout(mqttCompletionTimeout);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(mqttQualityOfService);
    return adapter;
}

@Bean
public IntegrationFlow myMqttInFlow() {
    return IntegrationFlows.from(mqttInbound)
            .handle(message -> {
                System.out.println(message);
            }).get();
}

更新：

也没用：

@Autowired
protected MessageHandler mqttOutbound;

@Test
public void test0() throws Exception {
    mqttOutbound.handleMessage(MessageBuilder.withPayload("test").build());
}


@SpringBootApplication
static class MqttSourceApplication {

    @Autowired
    private MqttPahoClientFactory mqttClientFactory;

    @Bean
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("test", mqttClientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic("test");
        messageHandler.setConverter(pahoMessageConverter());
        return messageHandler;
    }

    @Bean
    public DefaultPahoMessageConverter pahoMessageConverter() {
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(1, false, "UTF-8");
        return converter;
    }

}

更新

更简单...同样的错误：

@Autowired
MqttPahoClientFactory mqttPahoClientFactory;

private MessageHandler mqttOutbound;

@Before
public void setUp() throws Exception {
    MqttPahoMessageHandler messageHandler  = new MqttPahoMessageHandler(mqttClientId, mqttPahoClientFactory);
    messageHandler.setAsync(false);
    messageHandler.setDefaultTopic(mqttTopic);
    messageHandler.setConverter(new DefaultPahoMessageConverter());
    mqttOutbound = messageHandler;

}

@Test
public void test0() throws Exception {
    mqttOutbound.handleMessage(MessageBuilder.withPayload("test").build());
    Thread.sleep(10000L);
}

Answer 1

好的，已解决：Paho 显然关闭了我的连接，因为测试和主要都使用相同的客户端 ID。解决方案是按照此处的建议将 clientId 替换为 MqttAsyncClient.generateClientId()：

Spring 集成：MQTT 集成测试基础

Spring Integration: MQTT integration test basics

spring

spring-integration

paho

spring-integration-dsl

spring-integration-mqtt