Merge branch 'develop' into fix_issue_7666

freqtrade/rpc/api_server/api_ws.py

@@ -127,13 +127,6 @@ async def message_endpoint(
             except Exception as e:
                 logger.info(f"Consumer connection failed - {channel}: {e}")
                 logger.debug(e, exc_info=e)
-            finally:
-                await channel_manager.on_disconnect(ws)
-
-        else:
-            if channel:
-                await channel_manager.on_disconnect(ws)
-            await ws.close()
 
     except RuntimeError:
         # WebSocket was closed
@@ -144,4 +137,5 @@ async def message_endpoint(
         # Log tracebacks to keep track of what errors are happening
         logger.exception(e)
     finally:
-        await channel_manager.on_disconnect(ws)
+        if channel:
+            await channel_manager.on_disconnect(ws)

freqtrade/rpc/api_server/webserver.py

@@ -197,6 +197,7 @@ class ApiServer(RPCHandler):
                 # Get data from queue
                 message: WSMessageSchemaType = await async_queue.get()
                 logger.debug(f"Found message of type: {message.get('type')}")
+                async_queue.task_done()
                 # Broadcast it
                 await self._ws_channel_manager.broadcast(message)
         except asyncio.CancelledError:
@@ -210,6 +211,9 @@ class ApiServer(RPCHandler):
             # Disconnect channels and stop the loop on cancel
             await self._ws_channel_manager.disconnect_all()
             self._ws_loop.stop()
+            # Avoid adding more items to the queue if they aren't
+            # going to get broadcasted.
+            self._ws_queue = None
 
     def start_api(self):
         """

freqtrade/rpc/api_server/ws/channel.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import time
 from threading import RLock
 from typing import Any, Dict, List, Optional, Type, Union
 from uuid import uuid4
@@ -46,7 +47,7 @@ class WebSocketChannel:
         self._relay_task = asyncio.create_task(self.relay())
 
         # Internal event to signify a closed websocket
-        self._closed = False
+        self._closed = asyncio.Event()
 
         # Wrap the WebSocket in the Serializing class
         self._wrapped_ws = self._serializer_cls(self._websocket)
@@ -73,15 +74,26 @@ class WebSocketChannel:
         Add the data to the queue to be sent.
         :returns: True if data added to queue, False otherwise
         """
+
+        # This block only runs if the queue is full, it will wait
+        # until self.drain_timeout for the relay to drain the outgoing queue
+        # We can't use asyncio.wait_for here because the queue may have been created with a
+        # different eventloop
+        start = time.time()
+        while self.queue.full():
+            await asyncio.sleep(1)
+            if (time.time() - start) > self.drain_timeout:
+                return False
+
+        # If for some reason the queue is still full, just return False
         try:
-            await asyncio.wait_for(
-                self.queue.put(data),
-                timeout=self.drain_timeout
-            )
-            return True
-        except asyncio.TimeoutError:
+            self.queue.put_nowait(data)
+        except asyncio.QueueFull:
             return False
 
+        # If we got here everything is ok
+        return True
+
     async def recv(self):
         """
         Receive data on the wrapped websocket
@@ -99,14 +111,19 @@ class WebSocketChannel:
         Close the WebSocketChannel
         """
 
-        self._closed = True
+        try:
+            await self.raw_websocket.close()
+        except Exception:
+            pass
+
+        self._closed.set()
         self._relay_task.cancel()
 
     def is_closed(self) -> bool:
         """
         Closed flag
         """
-        return self._closed
+        return self._closed.is_set()
 
     def set_subscriptions(self, subscriptions: List[str] = []) -> None:
         """
@@ -129,7 +146,7 @@ class WebSocketChannel:
         Relay messages from the channel's queue and send them out. This is started
         as a task.
         """
-        while True:
+        while not self._closed.is_set():
             message = await self.queue.get()
             try:
                 await self._send(message)

freqtrade/rpc/external_message_consumer.py

@@ -264,10 +264,10 @@ class ExternalMessageConsumer:
                 # We haven't received data yet. Check the connection and continue.
                 try:
                     # ping
-                    ping = await channel.ping()
+                    pong = await channel.ping()
+                    latency = (await asyncio.wait_for(pong, timeout=self.ping_timeout) * 1000)
 
-                    await asyncio.wait_for(ping, timeout=self.ping_timeout)
-                    logger.debug(f"Connection to {channel} still alive...")
+                    logger.info(f"Connection to {channel} still alive, latency: {latency}ms")
 
                     continue
                 except (websockets.exceptions.ConnectionClosed):
@@ -276,7 +276,7 @@ class ExternalMessageConsumer:
                     await asyncio.sleep(self.sleep_time)
                     break
                 except Exception as e:
-                    logger.warning(f"Ping error {channel} - retrying in {self.sleep_time}s")
+                    logger.warning(f"Ping error {channel} - {e} - retrying in {self.sleep_time}s")
                     logger.debug(e, exc_info=e)
                     await asyncio.sleep(self.sleep_time)