Merge pull request #19 from dist-sys/add-authentication

nabon · web-flow · commit 2cf7b3214ffa · 2020-11-16T16:38:17.000+09:00
Add authentication
diff --git a/doc/usage_en.md b/doc/usage_en.md
@@ -1,4 +1,4 @@
-# MQTTLoader usage (v0.7.2)
+# MQTTLoader usage (v0.7.3)
 MQTTLoader is a load testing tool (client tool) for MQTT.  
 It supports both MQTT v5.0 and v3.1.1.
 
@@ -27,11 +27,11 @@ You can display the help by:
 `$ ./mqttloader -h`
 
 ```
-MQTTLoader version 0.7.0
+MQTTLoader version 0.7.3
 usage: mqttloader.Loader -b <arg> [-v <arg>] [-p <arg>] [-s <arg>] [-pq
        <arg>] [-sq <arg>] [-ss] [-r] [-t <arg>] [-d <arg>] [-m <arg>] [-ru
        <arg>] [-rd <arg>] [-i <arg>] [-st <arg>] [-et <arg>] [-l <arg>]
-       [-n <arg>] [-im] [-h]
+       [-n <arg>] [-im] [-un <arg>] [-pw <arg>] [-h]
  -b,--broker <arg>        Broker URL. E.g., tcp://127.0.0.1:1883
  -v,--version <arg>       MQTT version ("3" for 3.1.1 or "5" for 5.0).
   :
@@ -98,6 +98,8 @@ Please refer to **3. Parameteres of MQTTLoader** for more details of each parame
 | -l \<arg\> | INFO | Log level. <br>Valid values are `SEVERE`/`WARNING`/`INFO`/`ALL`. |
 | -n \<arg\> | (none) | URL of the NTP server. By setting this, time synchronization is enabled. <br>Ex. `ntp.nict.jp` |
 | -im \<arg\> | (none) | Run MQTTLoader by in-memory mode. By default, MQTTLoader writes out measurement records to a file. |
+| -un \<arg\> | (none) | User name. Required if the broker has the configuration of password authentication. |
+| -pw \<arg\> | (none) | Password. Required if the broker has the configuration of password authentication. |
 | -h |  | Display help. |
 
 MQTTLoader starts to terminate when all of the following conditions are met.  
diff --git a/doc/usage_jp.md b/doc/usage_jp.md
@@ -1,4 +1,4 @@
-# MQTTLoader 利用方法 (v0.7.2)
+# MQTTLoader 利用方法 (v0.7.3)
 MQTTLoaderは、MQTT v5.0とv3.1.1に対応した負荷テストツール（クライアントツール）です。
 
 ## 1. 環境要件
@@ -28,11 +28,11 @@ Windowsユーザはmqttloader.bat（バッチファイル）を、Linux等のユ
 `$ ./mqttloader -h`
 
 ```
-MQTTLoader version 0.7.0
+MQTTLoader version 0.7.3
 usage: mqttloader.Loader -b <arg> [-v <arg>] [-p <arg>] [-s <arg>] [-pq
        <arg>] [-sq <arg>] [-ss] [-r] [-t <arg>] [-d <arg>] [-m <arg>] [-ru
        <arg>] [-rd <arg>] [-i <arg>] [-st <arg>] [-et <arg>] [-l <arg>]
-       [-n <arg>] [-im] [-h]
+       [-n <arg>] [-im] [-un <arg>] [-pw <arg>] [-h]
  -b,--broker <arg>        Broker URL. E.g., tcp://127.0.0.1:1883
  -v,--version <arg>       MQTT version ("3" for 3.1.1 or "5" for 5.0).
   :
@@ -99,6 +99,8 @@ publisherとsubscriberを別マシンで動かすことで、負荷が相互に
 | -l \<arg\> | INFO | ログレベル。<br>設定可能な値：`SEVERE`/`WARNING`/`INFO`/`ALL` |
 | -n \<arg\> | （無し） | NTPサーバのURL。設定すると時刻同期が有効になる（デフォルト：無効）。<br>例：`ntp.nict.jp`　 |
 | -im \<arg\> | （無し） | MQTTLoaderをメモリ上でのみ動作させる。デフォルトでは、測定レコードはファイルに書き出される。 |
+| -un \<arg\> | （無し） | ユーザ名（ブローカにてパスワード認証が設定されている場合に指定）。 |
+| -pw \<arg\> | （無し） | パスワード（ブローカにてパスワード認証が設定されている場合に指定）。 |
 | -h |  | ヘルプを表示 |
 
 MQTTLoaderは、以下の条件をすべて満たすと、クライアントを切断させ終了します。  
diff --git a/src/main/java/mqttloader/Constants.java b/src/main/java/mqttloader/Constants.java
@@ -19,7 +19,7 @@
 import java.text.SimpleDateFormat;
 
 public class Constants {
-    public static final String VERSION = "0.7.2";
+    public static final String VERSION = "0.7.3";
     public static final String FILE_NAME_PREFIX = "mqttloader_";
     private static final String HOST_ID = Util.genRandomChars(4);
     public static final String SUB_CLIENT_ID_PREFIX = "ml-"+HOST_ID+"-s-";
@@ -51,6 +51,8 @@ public enum Opt {
         LOG_LEVEL("l", "log", true, "Log level (SEVERE/WARNING/INFO/ALL).", "INFO"),
         NTP("n", "ntp", true, "NTP server. E.g., ntp.nict.jp", null),
         IN_MEMORY("im", "inmemory", false, "Enable in-memory mode", null),
+        USERNAME("un", "username", true, "User name for authentication", null),
+        PASSWORD("pw", "password", true, "Password for authentication", null),
         HELP("h", "help", false, "Display help.", null);
 
         private String name;
diff --git a/src/main/java/mqttloader/Loader.java b/src/main/java/mqttloader/Loader.java
@@ -160,6 +160,8 @@ private void prepareClients() {
             broker = "tcp://"+broker;
         }
         int version = Util.getOptValInt(Opt.MQTT_VERSION);
+        String userName = Util.getOptVal(Opt.USERNAME);
+        String password = Util.getOptVal(Opt.PASSWORD);
         int numPub = Util.getOptValInt(Opt.NUM_PUB);
         int numSub = Util.getOptValInt(Opt.NUM_SUB);
         int pubQos = Util.getOptValInt(Opt.PUB_QOS);
@@ -172,17 +174,17 @@ private void prepareClients() {
         int pubInterval = Util.getOptValInt(Opt.INTERVAL);
         for(int i=0;i<numPub;i++){
             if(version==5){
-                publishers.add(new PublisherV5(i, broker, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));
+                publishers.add(new PublisherV5(i, broker, userName, password, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));
             }else{
-                publishers.add(new PublisherV3(i, broker, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));
+                publishers.add(new PublisherV3(i, broker, userName, password, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));
             }
         }
 
         for(int i=0;i<numSub;i++){
             if(version==5){
-                subscribers.add(new SubscriberV5(i, broker, subQos, shSub, topic, recorder));
+                subscribers.add(new SubscriberV5(i, broker, userName, password, subQos, shSub, topic, recorder));
             }else{
-                subscribers.add(new SubscriberV3(i, broker, subQos, topic, recorder));
+                subscribers.add(new SubscriberV3(i, broker, userName, password, subQos, topic, recorder));
             }
         }
     }
diff --git a/src/main/java/mqttloader/client/PublisherV3.java b/src/main/java/mqttloader/client/PublisherV3.java
@@ -16,6 +16,8 @@
 
 package mqttloader.client;
 
+import java.nio.charset.StandardCharsets;
+
 import mqttloader.Loader;
 import mqttloader.Recorder;
 import mqttloader.Util;
@@ -29,14 +31,16 @@ public class PublisherV3 extends AbstractPublisher {
     private MqttClient client;
     private MqttMessage message = new MqttMessage();
 
-    public PublisherV3(int clientNumber, String broker, int qos, boolean retain, String topic, int payloadSize, int numMessage, int pubInterval, Recorder recorder) {
+    public PublisherV3(int clientNumber, String broker, String userName, String password, int qos, boolean retain, String topic, int payloadSize, int numMessage, int pubInterval, Recorder recorder) {
         super(clientNumber, topic, payloadSize, numMessage, pubInterval, recorder);
         message.setQos(qos);
         message.setRetained(retain);
 
         MqttConnectOptions options = new MqttConnectOptions();
         options.setMqttVersion(4);
         options.setCleanSession(true);
+        if(userName != null) options.setUserName(userName);
+        if(password != null) options.setPassword(password.toCharArray());
         try {
             client = new MqttClient(broker, clientId, new MemoryPersistence());
             client.connect(options);
diff --git a/src/main/java/mqttloader/client/PublisherV5.java b/src/main/java/mqttloader/client/PublisherV5.java
@@ -16,6 +16,8 @@
 
 package mqttloader.client;
 
+import java.nio.charset.StandardCharsets;
+
 import mqttloader.Loader;
 import mqttloader.Recorder;
 import mqttloader.Util;
@@ -29,13 +31,15 @@ public class PublisherV5 extends AbstractPublisher {
     private MqttClient client;
     private MqttMessage message = new MqttMessage();
 
-    public PublisherV5(int clientNumber, String broker, int qos, boolean retain, String topic, int payloadSize, int numMessage, int pubInterval, Recorder recorder) {
+    public PublisherV5(int clientNumber, String broker, String userName, String password, int qos, boolean retain, String topic, int payloadSize, int numMessage, int pubInterval, Recorder recorder) {
         super(clientNumber, topic, payloadSize, numMessage, pubInterval, recorder);
         message.setQos(qos);
         message.setRetained(retain);
 
         MqttConnectionOptions options = new MqttConnectionOptions();
         options.setCleanStart(true);
+        if(userName != null) options.setUserName(userName);
+        if(password != null) options.setPassword(password.getBytes(StandardCharsets.UTF_8));
         try {
             client = new MqttClient(broker, clientId, new MemoryPersistence());
             client.connect(options);
diff --git a/src/main/java/mqttloader/client/SubscriberV3.java b/src/main/java/mqttloader/client/SubscriberV3.java
@@ -30,11 +30,13 @@
 public class SubscriberV3 extends AbstractSubscriber implements MqttCallback {
     private MqttClient client;
 
-    public SubscriberV3(int clientNumber, String broker, int qos, String topic, Recorder recorder) {
+    public SubscriberV3(int clientNumber, String broker, String userName, String password, int qos, String topic, Recorder recorder) {
         super(clientNumber, recorder);
         MqttConnectOptions options = new MqttConnectOptions();
         options.setMqttVersion(4);
         options.setCleanSession(true);
+        if(userName != null) options.setUserName(userName);
+        if(password != null) options.setPassword(password.toCharArray());
         try {
             client = new MqttClient(broker, clientId, new MemoryPersistence());
             client.setCallback(this);
diff --git a/src/main/java/mqttloader/client/SubscriberV5.java b/src/main/java/mqttloader/client/SubscriberV5.java
@@ -16,6 +16,8 @@
 
 package mqttloader.client;
 
+import java.nio.charset.StandardCharsets;
+
 import mqttloader.Loader;
 import mqttloader.Recorder;
 import org.eclipse.paho.mqttv5.client.IMqttToken;
@@ -31,10 +33,12 @@
 public class SubscriberV5 extends AbstractSubscriber implements MqttCallback {
     private MqttClient client;
 
-    public SubscriberV5(int clientNumber, String broker, int qos, boolean shSub, String topic, Recorder recorder) {
+    public SubscriberV5(int clientNumber, String broker, String userName, String password, int qos, boolean shSub, String topic, Recorder recorder) {
         super(clientNumber, recorder);
         MqttConnectionOptions options = new MqttConnectionOptions();
         options.setCleanStart(true);
+        if(userName != null) options.setUserName(userName);
+        if(password != null) options.setPassword(password.getBytes(StandardCharsets.UTF_8));
         try {
             client = new MqttClient(broker, clientId, new MemoryPersistence());
             client.setCallback(this);

Original file line number	Diff line number	Diff line change
`@@ -160,6 +160,8 @@ private void prepareClients() {`
`160`	`160`	`broker = "tcp://"+broker;`
`161`	`161`	`}`
`162`	`162`	`int version = Util.getOptValInt(Opt.MQTT_VERSION);`
	`163`	`+ String userName = Util.getOptVal(Opt.USERNAME);`
	`164`	`+ String password = Util.getOptVal(Opt.PASSWORD);`
`163`	`165`	`int numPub = Util.getOptValInt(Opt.NUM_PUB);`
`164`	`166`	`int numSub = Util.getOptValInt(Opt.NUM_SUB);`
`165`	`167`	`int pubQos = Util.getOptValInt(Opt.PUB_QOS);`
`@@ -172,17 +174,17 @@ private void prepareClients() {`
`172`	`174`	`int pubInterval = Util.getOptValInt(Opt.INTERVAL);`
`173`	`175`	`for(int i=0;i<numPub;i++){`
`174`	`176`	`if(version==5){`
`175`		`- publishers.add(new PublisherV5(i, broker, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));`
	`177`	`+ publishers.add(new PublisherV5(i, broker, userName, password, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));`
`176`	`178`	`}else{`
`177`		`- publishers.add(new PublisherV3(i, broker, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));`
	`179`	`+ publishers.add(new PublisherV3(i, broker, userName, password, pubQos, retain, topic, payloadSize, numMessage, pubInterval, recorder));`
`178`	`180`	`}`
`179`	`181`	`}`
`180`	`182`
`181`	`183`	`for(int i=0;i<numSub;i++){`
`182`	`184`	`if(version==5){`
`183`		`- subscribers.add(new SubscriberV5(i, broker, subQos, shSub, topic, recorder));`
	`185`	`+ subscribers.add(new SubscriberV5(i, broker, userName, password, subQos, shSub, topic, recorder));`
`184`	`186`	`}else{`
`185`		`- subscribers.add(new SubscriberV3(i, broker, subQos, topic, recorder));`
	`187`	`+ subscribers.add(new SubscriberV3(i, broker, userName, password, subQos, topic, recorder));`
`186`	`188`	`}`
`187`	`189`	`}`
`188`	`190`	`}`