Merge pull request #2171 from Drakkar-Software/dev

Master merge

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 *It is strongly advised to perform an update of your tentacles after updating OctoBot. (start.py tentacles --install --all)*
 
+## [0.4.32] - 2022-12-29
+### Fixed
+- MQTT reconnection issues
+
 ## [0.4.31] - 2022-12-28
 ### Fixed
 - profile duplicate download

README.md

@@ -1,4 +1,4 @@
-# OctoBot [0.4.31](https://octobot.click/gh-changelog)
+# OctoBot [0.4.32](https://octobot.click/gh-changelog)
 [![PyPI](https://img.shields.io/pypi/v/OctoBot.svg)](https://octobot.click/gh-pypi)
 [![Codacy Badge](https://api.codacy.com/project/badge/Grade/e07fb190156d4efb8e7d07aaa5eff2e1)](https://app.codacy.com/gh/Drakkar-Software/OctoBot?utm_source=github.com&utm_medium=referral&utm_content=Drakkar-Software/OctoBot&utm_campaign=Badge_Grade_Dashboard)[![Downloads](https://pepy.tech/badge/octobot/month)](https://pepy.tech/project/octobot)
 [![Dockerhub](https://img.shields.io/docker/pulls/drakkarsoftware/octobot.svg)](https://octobot.click/gh-dockerhub)

octobot/__init__.py

@@ -16,5 +16,5 @@
 
 PROJECT_NAME = "OctoBot"
 AUTHOR = "Drakkar-Software"
-VERSION = "0.4.31"  # major.minor.revision
+VERSION = "0.4.32"  # major.minor.revision
 LONG_VERSION = f"{VERSION}"

octobot/community/feeds/community_mqtt_feed.py

@@ -38,6 +38,7 @@ class CommunityMQTTFeed(abstract_feed.AbstractFeed):
     DISABLE_RECONNECT_VALUE = -2
     DEVICE_CREATE_TIMEOUT = 5 * commons_constants.MINUTE_TO_SECONDS
     DEVICE_CREATION_REFRESH_DELAY = 2
+    CONNECTION_TIMEOUT = 10
 
     # Quality of Service level determines the reliability of the data flow between a client and a message broker.
     # The message may be sent in three ways:
@@ -68,7 +69,16 @@ def __init__(self, feed_url, authenticator):
     async def start(self):
         self.should_stop = False
         self._device_uuid = self.authenticator.user_account.get_selected_bot_device_uuid()
-        await self._connect()
+        try:
+            await self._connect()
+            if self.is_connected():
+                self.logger.info("Successful connection request to mqtt device")
+            else:
+                self.logger.info("Failed to connect to mqtt device")
+        except asyncio.TimeoutError as err:
+            self.logger.exception(err, True, f"Timeout when connecting to mqtt device: {err}")
+        except Exception as err:
+            self.logger.exception(err, True, f"Unexpected error when connecting to mqtt device: {err}")
 
     async def stop(self):
         self.logger.debug("Stopping ...")
@@ -86,8 +96,8 @@ async def restart(self):
             if not self.should_stop:
                 await self.stop()
             await self.start()
-        except Exception as e:
-            self.logger.exception(e, True, f"Error when restarting mqtt feed: {e}")
+        except Exception as err:
+            self.logger.exception(err, True, f"{err}")
 
     def is_using_bot_device(self, user_account):
         try:
@@ -232,20 +242,23 @@ async def _reconnect(self, client):
                 await self._stop_mqtt_client()
             except Exception as e:
                 self.logger.debug(f"Ignored error while stopping client: {e}.")
-            first_reconnect = True
+            attempt = 1
             while not self.should_stop:
-                delay = 0 if first_reconnect else self.RECONNECT_DELAY
+                delay = 0 if attempt == 1 else self.RECONNECT_DELAY
+                error = None
                 try:
-                    self.logger.info(f"Reconnecting, client_id: {client._client_id}")
+                    self.logger.info(f"Reconnecting, client_id: {client._client_id} (attempt {attempt})")
                     await self._connect()
                     await asyncio.sleep(self.RECONNECT_ENSURE_DELAY)
                     if self.is_connected():
                         self.logger.info(f"Reconnected, client_id: {client._client_id}")
                         return
                     error = "failed to connect"
                 except Exception as err:
-                    error = err
-                first_reconnect = False
+                    error = f"{err}"
+                finally:
+                    self.logger.debug(f"Reconnect attempt {attempt} {'succeeded' if error is None else 'failed'}.")
+                    attempt += 1
                 self.logger.debug(f"Error while reconnecting: {error}. Trying again in {delay} seconds.")
                 await asyncio.sleep(delay)
         finally:
@@ -318,13 +331,17 @@ async def _connect(self):
             self._mqtt_client.connect(self.feed_url, self.mqtt_broker_port, version=self.MQTT_VERSION)
         )
         try:
-            await self._connect_task
+            await asyncio.wait_for(self._connect_task, self.CONNECTION_TIMEOUT)
             self._connected_at_least_once = True
         except asyncio.CancelledError:
             # got cancelled by on_disconnect, can't connect
             self.logger.error(f"Can't connect to server, make sure that your device uuid is valid. "
                               f"Current mqtt uuid is: {self._device_uuid}")
             self._valid_auth = False
+        except asyncio.TimeoutError as err:
+            message = "Timeout error when trying to connect to mqtt device"
+            self.logger.debug(message)
+            raise asyncio.TimeoutError(message) from err
 
     def _subscribe(self, topics):
         if not topics:

requirements.txt

@@ -6,7 +6,7 @@ OctoBot-Commons==1.8.2
 OctoBot-Trading==2.3.2
 OctoBot-Evaluators==1.8.0
 OctoBot-Tentacles-Manager==2.8.1
-OctoBot-Services==1.4.0
+OctoBot-Services==1.4.1
 OctoBot-Backtesting==1.8.0
 Async-Channel==2.1.0
 trading-backend==1.0.17