MqttPublisher.java

/*
 * Copyright (c) 2009, 2012 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *    Dave Locke - initial API and implementation and/or initial documentation
 */

/**
 * Modified with IntelliJ IDEA by
 * User: cygairko
 * Date: 02.05.2014
 * Time: 03:06
 */

package de.devcube.distribution.mqtools;

import de.devcube.distribution.help.Properties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

public class MqttPublisher implements MqttCallback {

    private static Logger logger = LogManager.getLogger(new Object() {}.getClass().getEnclosingClass().getSimpleName());

    // Private instance variables
    private static MqttClient client;
    private static MqttConnectOptions conOpt;
    private boolean quietMode;
    private String password;
    private String username;

    static {
        //This sample stores in a temporary directory... where messages temporarily
        // stored until the message has been delivered to the server.
        //..a real application ought to store them somewhere
        // where they are not likely to get deleted or tampered with
        String tmpDir = System.getProperty("java.io.tmpdir");
        MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
        try {
            // Construct the object that contains connection parameters
            // such as cleanSession and LWT
            conOpt = new MqttConnectOptions();
            conOpt.setCleanSession(true);

            // Set password and username if parameters are set in config file
            String password;
            String username;
            if (!(password = Properties.getConfig("MQTT_PASSWORD")).isEmpty()) {
                conOpt.setPassword(password.toCharArray());
            }
            if (!(username = Properties.getConfig("MQTT_USERNAME")).isEmpty()) {
                conOpt.setUserName(username);
            }

            client = new MqttClient(Properties.getConfig("MQTT_BROKER_URL"), Properties.getConfig("DEVICE_ID"), dataStore);
        } catch (MqttException me) {
            logger.error(me);
        }

    }

    public MqttPublisher() {

    }

    /**
     * Publish / send a message to an MQTT server
     *
     * @param topicName the name of the topic to publish to
     * @param payload   the set of bytes to send to the MQTT server
     * @throws MqttException
     */
    public static void publish(String topicName, byte[] payload) throws MqttException {

        // Connect to the MQTT server
        logger.debug("Connecting to " + client.getServerURI() + " with client ID " + client.getClientId());
        client.connect(conOpt);

        logger.debug("Publishing to topic \"" + topicName);

        // Create and configure a message
        MqttMessage message = new MqttMessage(payload);
        message.setQos(Integer.valueOf(Properties.getConfig("MQTT_QOS")));

        // Send the message to the server, control is not returned until
        // it has been delivered to the server meeting the specified
        // quality of service.
        client.publish(topicName, message);

        // Disconnect the client
        client.disconnect();
        logger.debug("Disconnected");
    }


    /****************************************************************
     * Methods to implement the MqttCallback interface
     ****************************************************************/

    /**
     * @see MqttCallback#connectionLost(Throwable)
     */
    public void connectionLost(Throwable cause) {
        // Called when the connection to the server has been lost.
        // An application may choose to implement reconnection
        // logic at this point. This sample simply exits.
        logger.debug("Connection lost! " + cause);
        System.exit(1);
    }

    /**
     * @see MqttCallback#deliveryComplete(IMqttDeliveryToken)
     */
    public void deliveryComplete(IMqttDeliveryToken token) {
        // Called when a message has been delivered to the
        // server. The token passed in here is the same one
        // that was passed to or returned from the original call to publish.
        // This allows applications to perform asynchronous
        // delivery without blocking until delivery completes.
        //
        // This sample demonstrates asynchronous deliver and
        // uses the token.waitForCompletion() call in the main thread which
        // blocks until the delivery has completed.
        // Additionally the deliveryComplete method will be called if
        // the callback is set on the client
        //
        // If the connection to the server breaks before delivery has completed
        // delivery of a message will complete after the client has re-connected.
        // The getPendingTokens method will provide tokens for any messages
        // that are still to be delivered.
    }

    /**
     * @see MqttCallback#messageArrived(String, MqttMessage)
     */
    public void messageArrived(String topic, MqttMessage message) throws MqttException {
        // Called when a message arrives from the server that matches any
        // subscription made by the client
        String incoming = new String(message.getPayload());
        logger.debug("new message:\t" + topic + "\t" + incoming);
    }

    /****************************************************************
     * End of MqttCallback methods
     ****************************************************************/
}