From 6098c70eee04d4622e6aebf0af6aa31ec7bd48e4 Mon Sep 17 00:00:00 2001 From: fusion44 Date: Sat, 10 Sep 2022 15:52:11 +0200 Subject: [PATCH] refactor: send_sse_message to broadcast_sse_msg --- app/main.py | 6 +++--- app/repositories/apps.py | 24 +++++++++++----------- app/repositories/bitcoin.py | 6 +++--- app/repositories/lightning.py | 14 ++++++------- app/repositories/ln_impl/lnd_grpc.py | 6 +++--- app/repositories/system.py | 8 ++++---- app/repositories/system_impl/raspiblitz.py | 6 +++--- app/sse_manager.py | 2 +- app/utils.py | 6 +++--- 9 files changed, 39 insertions(+), 39 deletions(-) diff --git a/app/main.py b/app/main.py index 1357745..ac4d702 100644 --- a/app/main.py +++ b/app/main.py @@ -38,7 +38,7 @@ from app.repositories.utils import ( get_full_client_warmup_data_bitcoinonly, ) from app.routers import apps, bitcoin, lightning, setup, system -from app.utils import SSE, send_sse_message, build_sse_event, sse_mgr +from app.utils import SSE, broadcast_sse_msg, build_sse_event, sse_mgr logging.basicConfig(level=logging.WARNING) @@ -87,7 +87,7 @@ async def on_startup(): await redis_plugin.init_app(app, config=config) await redis_plugin.init() register_cookie_updater() - await send_sse_message(SSE.SYSTEM_STARTUP_INFO, api_startup_status.dict()) + await broadcast_sse_msg(SSE.SYSTEM_STARTUP_INFO, api_startup_status.dict()) loop = asyncio.get_event_loop() loop.create_task(_initialize_bitcoin()) @@ -119,7 +119,7 @@ async def _set_startup_status( loop = asyncio.get_event_loop() loop.create_task(warmup_new_connections()) - await send_sse_message(SSE.SYSTEM_STARTUP_INFO, api_startup_status.dict()) + await broadcast_sse_msg(SSE.SYSTEM_STARTUP_INFO, api_startup_status.dict()) async def _initialize_bitcoin(): diff --git a/app/repositories/apps.py b/app/repositories/apps.py index c055c46..9f07ec5 100644 --- a/app/repositories/apps.py +++ b/app/repositories/apps.py @@ -10,7 +10,7 @@ from fastapi import HTTPException, status from fastapi.encoders import jsonable_encoder from app.constants import available_app_ids -from app.utils import SSE, call_sudo_script, parse_key_value_text, send_sse_message +from app.utils import SSE, broadcast_sse_msg, call_sudo_script, parse_key_value_text SHELL_SCRIPT_PATH = config("shell_script_path") @@ -153,7 +153,7 @@ async def install_app_sub(app_id: str): detail=app_id + "install script does not exist / is not supported", ) - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, {"id": app_id, "mode": "on", "result": "running", "details": ""}, ) @@ -170,7 +170,7 @@ async def uninstall_app_sub(app_id: str, delete_data: bool): status.HTTP_400_BAD_REQUEST, detail="script not exist/supported" ) - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, {"id": app_id, "mode": "off", "result": "running", "details": ""}, ) @@ -235,7 +235,7 @@ async def run_bonus_script(app_id: str, params: str): # when there is a defined error message (if multiple it wil lbe the last one) if "error" in stdoutData: logging.error(f"FOUND `error=` returned by script: {stdoutData['error']}") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -247,7 +247,7 @@ async def run_bonus_script(app_id: str, params: str): # when there is no result (e.g. result="OK") at the end of install script stdout - consider also script had error elif not "result" in stdoutData: logging.error(f"NO `result=` returned by script:") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -266,7 +266,7 @@ async def run_bonus_script(app_id: str, params: str): if updatedAppData["error"] != "": logging.warning(f"Error Detected ...") logging.warning(f"updatedAppData: {updatedAppData}") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -280,7 +280,7 @@ async def run_bonus_script(app_id: str, params: str): elif mode == "on": if updatedAppData["installed"]: logging.info(f"WIN - install was effective") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -295,7 +295,7 @@ async def run_bonus_script(app_id: str, params: str): logging.error(f"FAIL - was not installed") logging.warning(f"DEBUG - updatedAppData: {updatedAppData}") logging.warning(f"DEBUG - params: {params}") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -312,7 +312,7 @@ async def run_bonus_script(app_id: str, params: str): logging.error(f"FAIL - is still installed") logging.warning(f"DEBUG - updatedAppData: {updatedAppData}") logging.warning(f"DEBUG - params: {params}") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -323,7 +323,7 @@ async def run_bonus_script(app_id: str, params: str): ) else: logging.info(f"WIN - uninstall was effective") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, { "id": app_id, @@ -334,10 +334,10 @@ async def run_bonus_script(app_id: str, params: str): ) # send an updated state if that app - await send_sse_message(SSE.INSTALLED_APP_STATUS, [updatedAppData]) + await broadcast_sse_msg(SSE.INSTALLED_APP_STATUS, [updatedAppData]) else: logging.warning(f"Install Feedback Event: fail no stdout") - await send_sse_message( + await broadcast_sse_msg( SSE.INSTALL_APP, {"id": app_id, "mode": params, "result": "fail", "details": "no stdout"}, ) diff --git a/app/repositories/bitcoin.py b/app/repositories/bitcoin.py index 1933ffb..65b3d1b 100644 --- a/app/repositories/bitcoin.py +++ b/app/repositories/bitcoin.py @@ -18,7 +18,7 @@ from app.models.bitcoind import ( NetworkInfo, ) from app.repositories.bitcoin_utils import bitcoin_config, bitcoin_rpc_async -from app.utils import SSE, send_sse_message +from app.utils import SSE, broadcast_sse_msg _initialized = False @@ -136,7 +136,7 @@ async def handle_block_sub_redis(verbosity: int = 1) -> str: ) r = await bitcoin_rpc_async("getblock", [hash, verbosity]) - await send_sse_message(SSE.BTC_NEW_BLOC, r["result"]) + await broadcast_sse_msg(SSE.BTC_NEW_BLOC, r["result"]) async def register_bitcoin_zmq_sub(): @@ -157,7 +157,7 @@ async def _handle_gather_bitcoin_status(): if last_info != info: # only send data if anything has changed - await send_sse_message(SSE.BTC_INFO, info.dict()) + await broadcast_sse_msg(SSE.BTC_INFO, info.dict()) last_info = info await asyncio.sleep(2) diff --git a/app/repositories/lightning.py b/app/repositories/lightning.py index 1824e89..ded49aa 100644 --- a/app/repositories/lightning.py +++ b/app/repositories/lightning.py @@ -22,7 +22,7 @@ from app.models.lightning import ( SendCoinsResponse, ) from app.models.system import APIPlatform -from app.utils import SSE, redis_get, send_sse_message +from app.utils import SSE, broadcast_sse_msg, redis_get PLATFORM = config("platform", cast=str) @@ -211,13 +211,13 @@ async def _handle_info_listener(): info = await ln.get_ln_info_impl() if last_info != info: - await send_sse_message(SSE.LN_INFO, info.dict()) + await broadcast_sse_msg(SSE.LN_INFO, info.dict()) last_info = info info_lite = LightningInfoLite.from_lninfo(info) if last_info_lite != info_lite: - await send_sse_message(SSE.LN_INFO_LITE, info_lite.dict()) + await broadcast_sse_msg(SSE.LN_INFO_LITE, info_lite.dict()) last_info_lite = info_lite await asyncio.sleep(GATHER_INFO_INTERVALL) @@ -225,7 +225,7 @@ async def _handle_info_listener(): async def _handle_invoice_listener(): async for i in ln.listen_invoices(): - await send_sse_message(SSE.LN_INVOICE_STATUS, i.dict()) + await broadcast_sse_msg(SSE.LN_INVOICE_STATUS, i.dict()) _schedule_wallet_balance_update() @@ -245,11 +245,11 @@ async def _handle_forward_event_listener(): if len(_fwd_successes) > 0: l = _fwd_successes _fwd_successes = [] - await send_sse_message(SSE.LN_FORWARD_SUCCESSES, l) + await broadcast_sse_msg(SSE.LN_FORWARD_SUCCESSES, l) _schedule_wallet_balance_update() rev = await get_fee_revenue() - await send_sse_message(SSE.LN_FEE_REVENUE, rev.dict()) + await broadcast_sse_msg(SSE.LN_FEE_REVENUE, rev.dict()) _fwd_update_scheduled = False @@ -272,7 +272,7 @@ def _schedule_wallet_balance_update(): await asyncio.sleep(1.1) wb = await ln.get_wallet_balance_impl() if _CACHE["wallet_balance"] != wb: - await send_sse_message(SSE.WALLET_BALANCE, wb.dict()) + await broadcast_sse_msg(SSE.WALLET_BALANCE, wb.dict()) _CACHE["wallet_balance"] = wb _wallet_balance_update_scheduled = False diff --git a/app/repositories/ln_impl/lnd_grpc.py b/app/repositories/ln_impl/lnd_grpc.py index eb65e1c..be9e5d8 100644 --- a/app/repositories/ln_impl/lnd_grpc.py +++ b/app/repositories/ln_impl/lnd_grpc.py @@ -33,7 +33,7 @@ from app.models.lightning import ( SendCoinsResponse, WalletBalance, ) -from app.utils import SSE, config_get_hex_str, send_sse_message +from app.utils import SSE, broadcast_sse_msg, config_get_hex_str _lnd_connect_error_debug_msg = """ LND_GRPC: Unable to connect to LND. Possible reasons: @@ -496,7 +496,7 @@ async def send_coins_impl(input: SendCoinsInput) -> SendCoinsResponse: response = await _lnd_stub.SendCoins(r) r = SendCoinsResponse.from_lnd_grpc(response, input) - await send_sse_message(SSE.LN_ONCHAIN_PAYMENT_STATUS, r.dict()) + await broadcast_sse_msg(SSE.LN_ONCHAIN_PAYMENT_STATUS, r.dict()) return r except grpc.aio._call.AioRpcError as error: _check_if_locked() @@ -533,7 +533,7 @@ async def send_payment_impl( p = None async for response in _router_stub.SendPaymentV2(r): p = Payment.from_lnd_grpc(response) - await send_sse_message(SSE.LN_PAYMENT_STATUS, p.dict()) + await broadcast_sse_msg(SSE.LN_PAYMENT_STATUS, p.dict()) return p except grpc.aio._call.AioRpcError as error: _check_if_locked(error) diff --git a/app/repositories/system.py b/app/repositories/system.py index fff203f..793083d 100644 --- a/app/repositories/system.py +++ b/app/repositories/system.py @@ -7,10 +7,10 @@ from fastapi import HTTPException, Request, status from app.models.system import APIPlatform, ConnectionInfo, RawDebugLogData, SystemInfo from app.utils import ( SSE, + broadcast_sse_msg, call_script, call_sudo_script, parse_key_value_text, - send_sse_message, ) PLATFORM = config("platform", default=APIPlatform.RASPIBLITZ) @@ -123,9 +123,9 @@ async def get_connection_info() -> ConnectionInfo: async def shutdown(reboot: bool) -> bool: if reboot: - await send_sse_message(SSE.SYSTEM_REBOOT_NOTICE, {"reboot": True}) + await broadcast_sse_msg(SSE.SYSTEM_REBOOT_NOTICE, {"reboot": True}) else: - await send_sse_message(SSE.SYSTEM_SHUTDOWN_NOTICE, {"shutdown": True}) + await broadcast_sse_msg(SSE.SYSTEM_SHUTDOWN_NOTICE, {"shutdown": True}) return await shutdown_impl(reboot=reboot) @@ -172,7 +172,7 @@ async def _handle_gather_hardware_info(): while True: info = await get_hardware_info() if last_info != info: - await send_sse_message(SSE.HARDWARE_INFO, info) + await broadcast_sse_msg(SSE.HARDWARE_INFO, info) last_info = info await asyncio.sleep(HW_INFO_YIELD_TIME) diff --git a/app/repositories/system_impl/raspiblitz.py b/app/repositories/system_impl/raspiblitz.py index c0d28b2..11e3a4a 100644 --- a/app/repositories/system_impl/raspiblitz.py +++ b/app/repositories/system_impl/raspiblitz.py @@ -9,10 +9,10 @@ from app.models.system import APIPlatform, ConnectionInfo, SystemInfo from app.repositories.lightning import get_ln_info from app.utils import ( SSE, + broadcast_sse_msg, call_script, parse_key_value_text, redis_get, - send_sse_message, ) SHELL_SCRIPT_PATH = config("shell_script_path") @@ -89,9 +89,9 @@ async def shutdown_impl(reboot: bool) -> bool: if proc.returncode > 0: err = stderr.decode() if reboot: - await send_sse_message(SSE.SYSTEM_REBOOT_ERROR, {"error_message": err}) + await broadcast_sse_msg(SSE.SYSTEM_REBOOT_ERROR, {"error_message": err}) else: - await send_sse_message(SSE.SYSTEM_SHUTDOWN_ERROR, {"error_message": err}) + await broadcast_sse_msg(SSE.SYSTEM_SHUTDOWN_ERROR, {"error_message": err}) return False diff --git a/app/sse_manager.py b/app/sse_manager.py index 6f252d9..9a3fdf6 100644 --- a/app/sse_manager.py +++ b/app/sse_manager.py @@ -1,8 +1,8 @@ import asyncio from asyncio.log import logger from typing import Tuple -import async_timeout +import async_timeout from fastapi import Request from app.external.sse_starlette import EventSourceResponse, ServerSentEvent diff --git a/app/utils.py b/app/utils.py index 4631184..496dfbd 100644 --- a/app/utils.py +++ b/app/utils.py @@ -8,12 +8,12 @@ import re import time import warnings from typing import Dict, Optional -from app.sse_manager import SSEManager from fastapi.encoders import jsonable_encoder from fastapi_plugins import redis_plugin from app.external.sse_starlette import ServerSentEvent +from app.sse_manager import SSEManager sse_mgr = SSEManager() sse_mgr.setup() @@ -40,8 +40,8 @@ def build_sse_event(event: str, json_data: Optional[Dict]): ) -async def send_sse_message(event: str, json_data: Optional[Dict]): - """Send a message to any SSE connections +async def broadcast_sse_msg(event: str, json_data: Optional[Dict]): + """Broadcasts a message to all connected clients Parameters ----------