Fix the exit status of `tedge mqtt pub` on error (#434)

* Revert to the previous behaviour: exit when the TCP connection is lost

* Cargo fmt

* Fix the error status returned by `tedge mqtt sub` on error

* Fix error message

* Cargo fmt

Co-authored-by: Wenzek <diw@softwareag.com>

6 files changed, 20 insertions, 6 deletions

diff --git a/tedge/src/cli/mqtt/error.rs b/tedge/src/cli/mqtt/error.rs
index 320d9fee..fe209154 100644
--- a/tedge/src/cli/mqtt/error.rs
+++ b/tedge/src/cli/mqtt/error.rs
@@ -11,4 +11,7 @@ pub enum MqttError {
 
     #[error("The input QoS should be 0, 1, or 2")]
     InvalidQoSError,
+
+    #[error("MQTT connection error: {0}\n\nHint: Is MQTT server running?")]
+    ServerError(String),
 }
diff --git a/tedge/src/cli/mqtt/publish.rs b/tedge/src/cli/mqtt/publish.rs
index 5e7503cc..c1a1dfa1 100644
--- a/tedge/src/cli/mqtt/publish.rs
+++ b/tedge/src/cli/mqtt/publish.rs
@@ -38,6 +38,7 @@ fn publish(cmd: &MqttPublishCommand) -> Result<(), MqttError> {
     let (mut client, mut connection) = rumqttc::Client::new(options, DEFAULT_QUEUE_CAPACITY);
     let mut published = false;
     let mut acknowledged = false;
+    let mut any_error = None;
 
     client.publish(&cmd.topic, cmd.qos, retain_flag, payload)?;
 
@@ -63,10 +64,11 @@ fn publish(cmd: &MqttPublishCommand) -> Result<(), MqttError> {
                 }
             }
             Ok(Event::Incoming(Incoming::Disconnect)) => {
+                any_error = Some(MqttError::ServerError("Disconnected".to_string()));
                 break;
             }
             Err(err) => {
-                eprintln!("ERROR: {:?}", err);
+                any_error = Some(MqttError::ServerError(err.to_string()));
                 break;
             }
             _ => {}
@@ -80,5 +82,9 @@ fn publish(cmd: &MqttPublishCommand) -> Result<(), MqttError> {
     }
 
     client.disconnect()?;
-    Ok(())
+    if let Some(err) = any_error {
+        Err(err)
+    } else {
+        Ok(())
+    }
 }
diff --git a/tedge/src/cli/mqtt/subscribe.rs b/tedge/src/cli/mqtt/subscribe.rs
index bde16811..b7278c75 100644
--- a/tedge/src/cli/mqtt/subscribe.rs
+++ b/tedge/src/cli/mqtt/subscribe.rs
@@ -63,7 +63,12 @@ fn subscribe(cmd: &MqttSubscribeCommand) -> Result<(), MqttError> {
                 client.subscribe(cmd.topic.as_str(), cmd.qos).unwrap();
             }
             Err(err) => {
-                eprintln!("ERROR: {:?}", err);
+                let err_msg = err.to_string();
+                if err_msg.contains("I/O: Connection refused (os error 111)") {
+                    return Err(MqttError::ServerError(err_msg));
+                }
+
+                eprintln!("ERROR: {}", err_msg);
                 std::thread::sleep(std::time::Duration::from_secs(1));
             }
             _ => {}
diff --git a/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos0/run.py b/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos0/run.py
index d1ca578c..1d732b4a 100644
--- a/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos0/run.py
+++ b/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos0/run.py
@@ -29,7 +29,7 @@ class PySysTest(BaseTest):
 
         pub = self.startProcess(
             command=self.sudo,
-            arguments=[self.tedge, "--qos", "0", "mqtt", "pub", "atopic", "amessage"],
+            arguments=[self.tedge, "mqtt", "pub", "--qos", "0", "atopic", "amessage"],
             stdouterr="tedge_pub_fail",
             expectedExitStatus="==1",
         )
diff --git a/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos1/run.py b/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos1/run.py
index 586d8fbe..be8a1961 100644
--- a/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos1/run.py
+++ b/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos1/run.py
@@ -29,7 +29,7 @@ class PySysTest(BaseTest):
 
         pub = self.startProcess(
             command=self.sudo,
-            arguments=[self.tedge, "--qos", "1", "mqtt", "pub", "atopic", "amessage"],
+            arguments=[self.tedge, "mqtt", "pub", "--qos", "1", "atopic", "amessage"],
             stdouterr="tedge_pub_fail",
             expectedExitStatus="==1",
         )
diff --git a/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos2/run.py b/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos2/run.py
index 971c1959..de8f9f3a 100644
--- a/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos2/run.py
+++ b/tests/PySys/local_mqtt_pub_fail/local_mqtt_pub_fail_qos2/run.py
@@ -29,7 +29,7 @@ class PySysTest(BaseTest):
 
         pub = self.startProcess(
             command=self.sudo,
-            arguments=[self.tedge, "--qos", "2", "mqtt", "pub", "atopic", "amessage"],
+            arguments=[self.tedge, "mqtt", "pub", "--qos", "2", "atopic", "amessage"],
             stdouterr="tedge_pub_fail",
             expectedExitStatus="==1",
         )