GlcoinCore/blitz_api
1
0
Fork 0
mirror of https://github.com/fusion44/blitz_api.git synced 2026-08-17 12:26:11 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: send_sse_message to broadcast_sse_msg

Browse source
This commit is contained in:
fusion44 2022-09-10 15:52:11 +02:00
parent 69d8032900
commit 6098c70eee
No known key found for this signature in database
9 changed files with 39 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

6
app/main.py
View file

@ -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():

24
app/repositories/apps.py
View file

@ -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"},
)

6
app/repositories/bitcoin.py
View file

@ -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)

14
app/repositories/lightning.py
View file

@ -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

6
app/repositories/ln_impl/lnd_grpc.py
View file

@ -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)

8
app/repositories/system.py
View file

@ -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)

6
app/repositories/system_impl/raspiblitz.py
View file

@ -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

2
app/sse_manager.py
View file

@ -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

6
app/utils.py
View file

@ -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
----------