Is AWS IOT SDK V2 Is thread safe

[SudipShresthaOr]

Hi Team,

I am trying to implement the publish function as below.
Here the publish function starts new instance of _publish in a separate thread. Main reason for separate thread is because I wanted to implement waiting for the publish_future result and also implemented retires there.

Question is that is the self._mqtt_connection.publish thread safe? The way I have implemented is Ok? As my program will be calling the publish function many times.

def _publish(self, message_topic, message_json, max_retries=3, timeout_secs=5):
    retries = 0
    while retries < max_retries:
        try:
            publish_future, _ = self._mqtt_connection.publish(
                topic=message_topic,
                payload=message_json,
                qos=mqtt.QoS.AT_LEAST_ONCE)
            publish_future.result(timeout=timeout_secs)  # Wait with timeout
            return  # Exit the function on success
        except TimeoutError:
            self._logger.info(f"Publish operation timed out. Retry {retries + 1}/{max_retries}.")
        except Exception as e:
            self._logger.info(f"An error occurred: {e}. Retry {retries + 1}/{max_retries}.")
        
        retries += 1
        time.sleep(1)  # Optional backoff between retries
        
    self._logger.debug("Mqtt Max retries reached, giving up.")

def publish(self, message_topic, message_json):
    threading.Thread(target=self._publish, args=(message_topic, message_json,)).start()  

[bretambrose]

Yes it is thread-safe. Beyond that, a dedicated thread per publish might be overkill, but it's your choice. Note that your implementation can easily accidentally publish the same message many times since the future timeout has no affect on the mqtt client's attempt to deliver the message. The returned future is just a one-way signal from the client to the user.

[SudipShresthaOr]

Thank you @bretambrose for your prompt response.