content: package io.emqx.stream.client.mqtt;
  
  import java.io.IOException;
  import java.sql.Timestamp;
  
  import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
  import org.eclipse.paho.client.mqttv3.MqttCallback;
  import org.eclipse.paho.client.mqttv3.MqttClient;
  import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  import org.eclipse.paho.client.mqttv3.MqttException;
  import org.eclipse.paho.client.mqttv3.MqttMessage;
  
  /**
   * A sample application that demonstrates how to use the Paho MQTT v3.1 Client blocking API.
   *
   * It can be run from the command line in one of two modes:
   *  - as a publisher, sending a single message to a topic on the server
   *  - as a subscriber, listening for messages from the server
   *
   *  There are three versions of the sample that implement the same features
   *  but do so using using different programming styles:
   *  <ol>
   *  <li>Sample (this one) which uses the API which blocks until the operation completes</li>
   *  <li>SampleAsyncWait shows how to use the asynchronous API with waiters that block until
   *  an action completes</li>
   *  <li>SampleAsyncCallBack shows how to use the asynchronous API where events are
   *  used to notify the application when an action completes<li>
   *  </ol>
   *
   *  If the application is run with the -h parameter then info is displayed that
   *  describes all of the options / parameters.
   */
  public class Sample implements MqttCallback {
  
    /**
     * The main entry point of the sample.
     *
     * This method handles parsing of the arguments specified on the
     * command-line before performing the specified action.
     */
    public static void main(String[] args) {
  
      // Default settings:
      boolean quietMode 	= false;
      String action 		= "publish";
      String topic 		= "";
      String message 		= "Message from blocking Paho MQTTv3 Java client sample";
      int qos 			= 2;
      String broker 		= "172.16.0.4";
      int port 			= 1883;
      String clientId 	= null;
      String subTopic		= "Sample/#";
      String pubTopic 	= "Sample/Java/v3";
      boolean cleanSession = true;			// Non durable subscriptions
      boolean ssl = false;
      String password = null;
      String userName = null;
      // Parse the arguments -
      for (int i=0; i<args.length; i++) {
        // Check this is a valid argument
        if (args[i].length() == 2 && args[i].startsWith("-")) {
          char arg = args[i].charAt(1);
          // Handle arguments that take no-value
          switch(arg) {
            case 'h': case '?':	printHelp(); return;
            case 'q': quietMode = true;	continue;
          }
  
          // Now handle the arguments that take a value and
          // ensure one is specified
          if (i == args.length -1 || args[i+1].charAt(0) == '-') {
            System.out.println("Missing value for argument: "+args[i]);
            printHelp();
            return;
          }
          switch(arg) {
            case 'a': action = args[++i];                 break;
            case 't': topic = args[++i];                  break;
            case 'm': message = args[++i];                break;
            case 's': qos = Integer.parseInt(args[++i]);  break;
            case 'b': broker = args[++i];                 break;
            case 'p': port = Integer.parseInt(args[++i]); break;
            case 'i': clientId = args[++i];				  break;
            case 'c': cleanSession = Boolean.valueOf(args[++i]);  break;
            case 'k': System.getProperties().put("javax.net.ssl.keyStore", args[++i]); break;
            case 'w': System.getProperties().put("javax.net.ssl.keyStorePassword", args[++i]); break;
            case 'r': System.getProperties().put("javax.net.ssl.trustStore", args[++i]); break;
            case 'v': ssl = Boolean.valueOf(args[++i]); break;
            case 'u': userName = args[++i];               break;
            case 'z': password = args[++i];               break;
            default:
              System.out.println("Unrecognised argument: "+args[i]);
              printHelp();
              return;
          }
        } else {
          System.out.println("Unrecognised argument: "+args[i]);
          printHelp();
          return;
        }
      }
  
      // Validate the provided arguments
      if (!action.equals("publish") && !action.equals("subscribe")) {
        System.out.println("Invalid action: "+action);
        printHelp();
        return;
      }
      if (qos < 0 || qos > 2) {
        System.out.println("Invalid QoS: "+qos);
        printHelp();
        return;
      }
      if (topic.equals("")) {
        // Set the default topic according to the specified action
        if (action.equals("publish")) {
          topic = pubTopic;
        } else {
          topic = subTopic;
        }
      }
  
      String protocol = "tcp://";
  
      if (ssl) {
        protocol = "ssl://";
      }
  
      String url = protocol + broker + ":" + port;
  
      if (clientId == null || clientId.equals("")) {
        clientId = "SampleJavaV3_"+action;
      }
  
      // With a valid set of arguments, the real work of
      // driving the client API can begin
      try {
        // Create an instance of this class
        Sample sampleClient = new Sample(url, clientId, cleanSession, quietMode,userName,password);
  
        // Perform the requested action
        if (action.equals("publish")) {
          sampleClient.publish(topic,qos,message.getBytes());
        } else {
          sampleClient.subscribe(topic,qos);
        }
      } catch(MqttException me) {
        // Display full details of any exception that occurs
        System.out.println("reason "+me.getReasonCode());
        System.out.println("msg "+me.getMessage());
        System.out.println("loc "+me.getLocalizedMessage());
        System.out.println("cause "+me.getCause());
        System.out.println("excep "+me);
        me.printStackTrace();
      }
    }
  
    // Private instance variables
    private MqttClient 			client;
    private final String 				brokerUrl;
    private final boolean 			quietMode;
    private MqttConnectOptions 	conOpt;
  
    /**
     * Constructs an instance of the sample client wrapper
     * @param brokerUrl the url of the server to connect to
     * @param clientId the client id to connect with
     * @param cleanSession clear state at end of connection or not (durable or non-durable subscriptions)
     * @param quietMode whether debug should be printed to standard out
     * @param userName the username to connect with
     * @param password the password for the user
     */
    private Sample(String brokerUrl, String clientId, boolean cleanSession, boolean quietMode, String userName, String password) {
      this.brokerUrl = brokerUrl;
      this.quietMode = quietMode;
      //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 connection options object that contains connection parameters
        // such as cleanSession and LWT
        conOpt = new MqttConnectOptions();
        conOpt.setCleanSession(cleanSession);
        if(password != null ) {
          conOpt.setPassword(password.toCharArray());
        }
        if(userName != null) {
          conOpt.setUserName(userName);
        }
  
        // Construct an MQTT blocking mode client
        try {
          client = new MqttClient(this.brokerUrl,clientId);
        }catch(Exception exp) {
          exp.printStackTrace();
          throw exp;
        }
  
  
        // Set this wrapper as the callback handler
        client.setCallback(this);
  
      } catch (MqttException e) {
        e.printStackTrace();
        log("Unable to set up client: "+e.toString());
        System.exit(1);
      }
    }
  
    /**
     * Publish / send a message to an MQTT server
     * @param topicName the name of the topic to publish to
     * @param qos the quality of service to delivery the message at (0,1,2)
     * @param payload the set of bytes to send to the MQTT server
     */
    private void publish(String topicName, int qos, byte[] payload) throws MqttException {
  
      try {
        log("Connecting to "+brokerUrl + " with client ID "+client.getClientId());
        client.connect(conOpt);
        log("Connected");
      }catch(Exception exp) {
        exp.printStackTrace();
        throw exp;
      }
  
  
      String time = new Timestamp(System.currentTimeMillis()).toString();
      log("Publishing at: "+time+ " to topic \""+topicName+"\" qos "+qos);
  
      // Create and configure a message
      MqttMessage message = new MqttMessage(payload);
      message.setQos(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();
      log("Disconnected");
    }
  
    /**
     * Subscribe to a topic on an MQTT server
     * Once subscribed this method waits for the messages to arrive from the server
     * that match the subscription. It continues listening for messages until the enter key is
     * pressed.
     * @param topicName to subscribe to (can be wild carded)
     * @param qos the maximum quality of service to receive messages at for this subscription
     */
    private void subscribe(String topicName, int qos) throws MqttException {
  
      // Connect to the MQTT server
      client.connect(conOpt);
      log("Connected to "+brokerUrl+" with client ID "+client.getClientId());
  
      // Subscribe to the requested topic
      // The QoS specified is the maximum level that messages will be sent to the client at.
      // For instance if QoS 1 is specified, any messages originally published at QoS 2 will
      // be downgraded to 1 when delivering to the client but messages published at 1 and 0
      // will be received at the same level they were published at.
      log("Subscribing to topic \""+topicName+"\" qos "+qos);
      client.subscribe(topicName, qos);
  
      // Continue waiting for messages until the Enter is pressed
      log("Press <Enter> to exit");
      try {
        //noinspection ResultOfMethodCallIgnored
        System.in.read();
      } catch (IOException e) {
        //If we can't read we'll just exit
      }
  
      // Disconnect the client from the server
      client.disconnect();
      log("Disconnected");
    }
  
    /**
     * Utility method to handle logging. If 'quietMode' is set, this method does nothing
     * @param message the message to log
     */
    private void log(String message) {
      if (!quietMode) {
        System.out.println(message);
      }
    }
  
    /* 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.
      log("Connection to " + brokerUrl + " 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) {
      // Called when a message arrives from the server that matches any
      // subscription made by the client
      String time = new Timestamp(System.currentTimeMillis()).toString();
      System.out.println("Time:\t" +time +
              "  Topic:\t" + topic +
              "  Message:\t" + new String(message.getPayload()) +
              "  QoS:\t" + message.getQos());
    }
  
    /* End of MqttCallback methods                                  */
    /****************************************************************/
  
    private static void printHelp() {
      System.out.println(
              "Syntax:\n\n" +
                      "    Sample [-h] [-a publish|subscribe] [-t <topic>] [-m <message text>]\n" +
                      "            [-s 0|1|2] -b <hostname|IP address>] [-p <brokerport>] [-i <clientID>]\n\n" +
                      "    -h  Print this help text and quit\n" +
                      "    -q  Quiet mode (default is false)\n" +
                      "    -a  Perform the relevant action (default is publish)\n" +
                      "    -t  Publish/subscribe to <topic> instead of the default\n" +
                      "            (publish: \"Sample/Java/v3\", subscribe: \"Sample/#\")\n" +
                      "    -m  Use <message text> instead of the default\n" +
                      "            (\"Message from MQTTv3 Java client\")\n" +
                      "    -s  Use this QoS instead of the default (2)\n" +
                      "    -b  Use this name/IP address instead of the default (m2m.eclipse.org)\n" +
                      "    -p  Use this port instead of the default (1883)\n\n" +
                      "    -i  Use this client ID instead of SampleJavaV3_<action>\n" +
                      "    -c  Connect to the server with a clean session (default is false)\n" +
                      "     \n\n Security Options \n" +
                      "     -u Username \n" +
                      "     -z Password \n" +
                      "     \n\n SSL Options \n" +
                      "    -v  SSL enabled; true - (default is false) " +
                      "    -k  Use this JKS format key store to verify the client\n" +
                      "    -w  Passpharse to verify certificates in the keys store\n" +
                      "    -r  Use this JKS format keystore to verify the server\n" +
                      " If javax.net.ssl properties have been set only the -v flag needs to be set\n" +
                      "Delimit strings containing spaces with \"\"\n\n" +
                      "Publishers transmit a single message then disconnect from the server.\n" +
                      "Subscribers remain connected to the server and receive appropriate\n" +
                      "messages until <enter> is pressed.\n\n"
      );
    }
  
  }