public connectAndSubscribe(String topic, IMqttMessageListener messageListener) throws MqttException {
    MqttClient client = new MqttClient(mosquittoConfig.getBrokerURI(), "testClientId");
    connectClientIfNeeded(client);
    MqttSubscription subscription = new MqttSubscription(topic, mosquittoConfig.getDefaultQos());
    return client.subscribe(new MqttSubscription[]{subscription}, new IMqttMessageListener[]{messageListener});
  }

  private void connectClientIfNeeded(MqttClient client) throws MqttException {
    if (!client.isConnected()) {
      client.connect(getConnectOptions());
    }
  }

  private MqttConnectionOptions getConnectOptions() {
    MqttConnectionOptions options = new MqttConnectionOptions();
    options.setCleanStart(false);
    options.setUserName(mosquittoConfig.getUser());
    options.setPassword(mosquittoConfig.getPassword().getBytes(StandardCharsets.UTF_8));
    options.setAutomaticReconnect(true);
    options.setRequestProblemInfo(true);
    options.setRequestResponseInfo(true);
    return options;
  }