public static void main(String... args) throws Throwable {
		final CountDownLatch connectLatch = new CountDownLatch(1);
		final AtomicReference<ConnectReturnCode> connectReturnCode = new AtomicReference<ConnectReturnCode>();
		AsyncClientListener listener = new AsyncClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				log.warn("Received a message when no subscriptions were active. Check your broker ;)");
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
				if (cause != null) {
					log.error("Disconnected from the broker due to an exception.", cause);
				} else {
					log.info("Disconnected from the broker.");
				}

				if (reconnecting) {
					log.info("Attempting to reconnect to the broker.");
				}
			}

			@Override
			public void connected(MqttClient client, ConnectReturnCode returnCode) {
				connectReturnCode.set(returnCode);
				connectLatch.countDown();
			}

			@Override
			public void subscribed(MqttClient client, Subscription[] requestedSubscriptions, Subscription[] grantedSubscriptions, boolean requestsGranted) {
			}

			@Override
			public void unsubscribed(MqttClient client, String[] topics) {
			}

			@Override
			public void published(MqttClient client, PublishMessage message) {
			}

		};

		// Build your client. This client is an asynchronous one so all interaction with the broker will be non-blocking.
		MqttClient client = new AsyncMqttClient("tcp://mqtt-broker:1883", listener, 5);
		try {
			// Connect to the broker. We will await the return code so that we know whether or not we can even begin publishing.
			client.connect("musicProducerAsync", false, "music-user", "music-pass");
			connectLatch.await();

			ConnectReturnCode returnCode = connectReturnCode.get();
			if (returnCode == null || returnCode != ConnectReturnCode.ACCEPTED) {
				// The broker bounced us. We are done.
				log.error("The broker rejected our attempt to connect. Reason: " + returnCode);
				return;
			}

			// Publish a musical catalog
			client.publish(new PublishMessage("grand/funk/railroad", QoS.AT_MOST_ONCE, "On Time"));
			client.publish(new PublishMessage("grand/funk/railroad", QoS.AT_MOST_ONCE, "E Pluribus Funk"));
			client.publish(new PublishMessage("jefferson/airplane", QoS.AT_MOST_ONCE, "Surrealistic Pillow"));
			client.publish(new PublishMessage("jefferson/airplane", QoS.AT_MOST_ONCE, "Crown of Creation"));
			client.publish(new PublishMessage("seventies/prog/rush", QoS.AT_MOST_ONCE, "2112"));
			client.publish(new PublishMessage("seventies/prog/rush", QoS.AT_MOST_ONCE, "A Farewell to Kings"));
			client.publish(new PublishMessage("seventies/prog/rush", QoS.AT_MOST_ONCE, "Hemispheres"));
		} catch (Exception ex) {
			log.error("An exception prevented the publishing of the full catalog.", ex);
		} finally {
			// We are done. Disconnect.
			if (!client.isClosed()) {
				client.disconnect();
			}
		}
	}