title: Spring Integration: MQTT integration test basics
body: <p>I try to test a simple MQTT listener created with Spring Boot and Spring Integration, but don't get it working. I've tried many approaches. The most promising was:</p>
  
  <pre><code>@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);
  }
  }
  </code></pre>
  
  <p>However, the test runs with 
  <code>2018-07-12 16:53:50.937 ERROR 21160 --- [T Rec: consumer] .m.i.MqttPahoMessageDrivenChannelAdapter : Lost connection: Verbindung wurde getrennt; retrying...</code>
  , and I don't see anything printed out.
  The code is mostly from: <a href="https://github.com/spring-projects/spring-integration-samples/tree/master/basic/mqtt" rel="nofollow noreferrer">https://github.com/spring-projects/spring-integration-samples/tree/master/basic/mqtt</a>
  . The example works nicely, but I need to be able to write proper integration tests.  </p>
  
  <p>The configuration is:</p>
  
  <pre><code>@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 -&gt; {
                  System.out.println(message);
              }).get();
  }
  </code></pre>
  
  <p><strong>UPDATE:</strong></p>
  
  <p>Didn't work either:</p>
  
  <pre><code>@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;
      }
  
  }
  </code></pre>
  
  <p><strong>UPDATE</strong></p>
  
  <p>Even simpler ... same error:</p>
  
  <pre><code>@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);
  }
  </code></pre>