fix: Ensure Unsubscribe waits for operation completion using getTokenError (#421)

Signed-off-by: FelixTing <[email protected]>

Author: FelixTing

Jenkinsfile

@@ -1,5 +1,6 @@
 //
 // Copyright (c) 2019 Intel Corporation
+// Copyright (c) 2025 IOTech Ltd
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,5 +16,6 @@
 //
 
 edgeXBuildGoMod (
-    project: 'go-mod-messaging'
+    project: 'go-mod-messaging',
+    goVersion: '1.23'
 )

internal/pkg/mqtt/client.go

@@ -201,8 +201,12 @@ func (mc *Client) Unsubscribe(topics ...string) error {
 	defer mc.subscriptionMutex.Unlock()
 
 	token := mc.mqttClient.Unsubscribe(topics...)
-	if token.Error() != nil {
-		return token.Error()
+
+	optionsReader := mc.mqttClient.OptionsReader()
+
+	err := getTokenError(token, optionsReader.WriteTimeout(), UnsubscribeOperation, "Failed to unsubscribe")
+	if err != nil {
+		return err
 	}
 
 	for _, topic := range topics {

internal/pkg/mqtt/errors.go

@@ -1,5 +1,6 @@
 /********************************************************************************
  *  Copyright 2020 Dell Inc.
+ *  Copyright (c) 2025 IOTech Ltd
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
  * in compliance with the License. You may obtain a copy of the License at
@@ -20,9 +21,10 @@ import (
 
 const (
 	// Different Client operations.
-	PublishOperation   = "Publish"
-	SubscribeOperation = "Subscribe"
-	ConnectOperation   = "Connect"
+	PublishOperation     = "Publish"
+	SubscribeOperation   = "Subscribe"
+	ConnectOperation     = "Connect"
+	UnsubscribeOperation = "Unsubscribe"
 )
 
 // TimeoutErr defines an error representing operations which have not completed and surpassed the allowed wait time.