adelorenzo/stable
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

emc: Fix potential startup timing issue

Browse Source
This commit is contained in:
Matthias 2022-09-12 19:50:22 +02:00
parent a477b3c244
commit 0052e58917
2 changed files with 7 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
freqtrade/rpc/external_message_consumer.py
View File

@ -15,6 +15,7 @@ from pydantic import ValidationError
from freqtrade.data.dataprovider import DataProvider from freqtrade.data.dataprovider import DataProvider
from freqtrade.enums import RPCMessageType from freqtrade.enums import RPCMessageType
from freqtrade.exceptions import OperationalException
from freqtrade.misc import remove_entry_exit_signals from freqtrade.misc import remove_entry_exit_signals
from freqtrade.rpc.api_server.ws.channel import WebSocketChannel from freqtrade.rpc.api_server.ws.channel import WebSocketChannel
from freqtrade.rpc.api_server.ws_schemas import (WSAnalyzedDFMessage, WSAnalyzedDFRequest, from freqtrade.rpc.api_server.ws_schemas import (WSAnalyzedDFMessage, WSAnalyzedDFRequest,
@ -57,7 +58,7 @@ class ExternalMessageConsumer:
self.producers = self._emc_config.get('producers', []) self.producers = self._emc_config.get('producers', [])
if self.enabled and len(self.producers) < 1: if self.enabled and len(self.producers) < 1:
raise ValueError("You must specify at least 1 Producer to connect to.") raise OperationalException("You must specify at least 1 Producer to connect to.")
self.wait_timeout = self._emc_config.get('wait_timeout', 300) # in seconds self.wait_timeout = self._emc_config.get('wait_timeout', 300) # in seconds
self.ping_timeout = self._emc_config.get('ping_timeout', 10) # in seconds self.ping_timeout = self._emc_config.get('ping_timeout', 10) # in seconds
@ -97,9 +98,8 @@ class ExternalMessageConsumer:
self._loop = asyncio.new_event_loop() self._loop = asyncio.new_event_loop()
self._thread = Thread(target=self._loop.run_forever) self._thread = Thread(target=self._loop.run_forever)
self._thread.start()
self._running = True self._running = True
self._thread.start()
self._main_task = asyncio.run_coroutine_threadsafe(self._main(), loop=self._loop) self._main_task = asyncio.run_coroutine_threadsafe(self._main(), loop=self._loop)

9
tests/rpc/test_rpc_emc.py
View File

@ -11,6 +11,7 @@ import pytest
import websockets import websockets
from freqtrade.data.dataprovider import DataProvider from freqtrade.data.dataprovider import DataProvider
from freqtrade.exceptions import OperationalException
from freqtrade.rpc.external_message_consumer import ExternalMessageConsumer from freqtrade.rpc.external_message_consumer import ExternalMessageConsumer
from tests.conftest import log_has, log_has_re, log_has_when from tests.conftest import log_has, log_has_re, log_has_when
@ -71,7 +72,7 @@ def test_emc_shutdown(patched_emc, caplog):
assert not log_has("Stopping ExternalMessageConsumer", caplog) assert not log_has("Stopping ExternalMessageConsumer", caplog)
def test_emc_init(patched_emc, default_conf, mocker, caplog): def test_emc_init(patched_emc, default_conf):
# Test the settings were set correctly # Test the settings were set correctly
assert patched_emc.initial_candle_limit <= 1500 assert patched_emc.initial_candle_limit <= 1500
assert patched_emc.wait_timeout > 0 assert patched_emc.wait_timeout > 0
@ -84,12 +85,10 @@ def test_emc_init(patched_emc, default_conf, mocker, caplog):
} }
}) })
dataprovider = DataProvider(default_conf, None, None, None) dataprovider = DataProvider(default_conf, None, None, None)
with pytest.raises(ValueError) as exc: with pytest.raises(OperationalException,
match="You must specify at least 1 Producer to connect to."):
ExternalMessageConsumer(default_conf, dataprovider) ExternalMessageConsumer(default_conf, dataprovider)
# Make sure we failed because of no producers
assert str(exc.value) == "You must specify at least 1 Producer to connect to."
# Parametrize this? # Parametrize this?
def test_emc_handle_producer_message(patched_emc, caplog, ohlcv_history): def test_emc_handle_producer_message(patched_emc, caplog, ohlcv_history):