ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-06-05 21:20:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

server: set event loop before initialising the state manager

Browse Source 
As the server is run first, we don't have to check if any other event
loop is running.
This commit is contained in:
Mose Müller 2025-05-22 15:57:09 +02:00
parent 9ee498eb5c
commit fb6ec16bf5
1 changed files with 4 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/pydase/server/server.py
View File

@ -14,7 +14,6 @@ from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager from pydase.data_service.state_manager import StateManager
from pydase.server.web_server import WebServer from pydase.server.web_server import WebServer
from pydase.task.autostart import autostart_service_tasks from pydase.task.autostart import autostart_service_tasks
from pydase.utils.helpers import current_event_loop_exists
HANDLED_SIGNALS = ( HANDLED_SIGNALS = (
signal.SIGINT, # Unix signal 2. Sent by Ctrl+C. signal.SIGINT, # Unix signal 2. Sent by Ctrl+C.
@ -162,6 +161,10 @@ class Server:
self._additional_servers = additional_servers self._additional_servers = additional_servers
self.should_exit = False self.should_exit = False
self.servers: dict[str, asyncio.Future[Any]] = {} self.servers: dict[str, asyncio.Future[Any]] = {}
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._state_manager = StateManager( self._state_manager = StateManager(
service=self._service, service=self._service,
filename=filename, filename=filename,
@ -170,11 +173,6 @@ class Server:
self._observer = DataServiceObserver(self._state_manager) self._observer = DataServiceObserver(self._state_manager)
self._state_manager.load_state() self._state_manager.load_state()
autostart_service_tasks(self._service) autostart_service_tasks(self._service)
if not current_event_loop_exists():
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
else:
self._loop = asyncio.get_event_loop()
def run(self) -> None: def run(self) -> None:
""" """