Added ctrl+c support

This commit is contained in:
Sven Heidemann 2021-11-17 19:35:42 +01:00
parent 5afbdb9b82
commit 5b61de0bf1
2 changed files with 30 additions and 9 deletions

View File

@ -1,8 +1,8 @@
from cpl_core.application import ApplicationABC
from cpl_core.configuration import ConfigurationABC
from cpl_core.console import Console
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from gismo_core.abc.bot_service_abc import BotServiceABC
from gismo_core.service.bot_service import BotService
@ -22,6 +22,15 @@ class Application(ApplicationABC):
try:
self._logger.trace(__name__, f'Try to start {BotService}')
await self._bot.start_async()
self._logger.trace(__name__, f'Stopped {BotService}')
except Exception as e:
self._logger.error(__name__, 'Start failed', e)
async def stop_async(self):
try:
self._logger.trace(__name__, f'Try to stop {BotService}')
await self._bot.close()
self._logger.trace(__name__, f'Stopped {BotService}')
except Exception as e:
self._logger.error(__name__, 'stop failed', e)
Console.write_line()

View File

@ -1,4 +1,5 @@
import asyncio
from typing import Optional
from cpl_core.application import ApplicationABC, ApplicationBuilder
@ -6,14 +7,25 @@ from gismo.application import Application
from gismo.startup import Startup
from modules.boot_log.boot_log_extension import BootLogExtension
class Main:
def __init__(self):
self._app: Optional[ApplicationABC] = None
async def main():
app_builder = ApplicationBuilder(Application)
app_builder.use_extension(BootLogExtension)
app_builder.use_startup(Startup)
app: ApplicationABC = await app_builder.build_async()
await app.run_async()
async def main(self):
app_builder = ApplicationBuilder(Application)
app_builder.use_extension(BootLogExtension)
app_builder.use_startup(Startup)
self._app: ApplicationABC = await app_builder.build_async()
await self._app.run_async()
async def stop(self):
await self._app.stop_async()
if __name__ == '__main__':
main = Main()
ml = asyncio.get_event_loop()
ml.run_until_complete(main())
try:
ml.run_until_complete(main.main())
except KeyboardInterrupt:
ml.run_until_complete(main.stop())