From 9dfd95032a7e02c07a010f5786a8a78693a03d3c Mon Sep 17 00:00:00 2001 From: Stefan Stammberger Date: Tue, 7 Sep 2021 20:59:23 +0200 Subject: [PATCH] refactor: move code from lnd impl to repository This avoids duplicated code with other ln implementations later on. This also pushes LightningStatus object to the SSE channel instead of the full LnInfo object. refs #11 --- app/repositories/lightning.py | 53 +++++++++++++++++++++----- app/repositories/ln_impl/clightning.py | 2 +- app/repositories/ln_impl/lnd.py | 52 +------------------------ app/utils.py | 1 + 4 files changed, 48 insertions(+), 60 deletions(-) diff --git a/app/repositories/lightning.py b/app/repositories/lightning.py index f393fef..5aae485 100644 --- a/app/repositories/lightning.py +++ b/app/repositories/lightning.py @@ -1,5 +1,8 @@ +import asyncio + from app.models.lightning import Invoice, LightningStatus, LnInfo, Payment from app.utils import SSE, lightning_config, send_sse_message +from decouple import config if lightning_config.ln_node == "lnd": from app.repositories.ln_impl.lnd import ( @@ -7,7 +10,7 @@ if lightning_config.ln_node == "lnd": get_implementation_name, get_ln_info_impl, get_wallet_balance_impl, - register_lightning_listener_impl, + listen_invoices, send_payment_impl, ) else: @@ -16,10 +19,14 @@ else: get_implementation_name, get_ln_info_impl, get_wallet_balance_impl, - register_lightning_listener_impl, + listen_invoices, send_payment_impl, ) +GATHER_INFO_INTERVALL = config("gather_ln_info_interval", default=5, cast=float) + +_CACHE = {"wallet_balance": None} + async def get_ln_status() -> LightningStatus: ln_info = await get_ln_info_impl() @@ -41,18 +48,46 @@ async def send_payment( pay_req: str, timeout_seconds: int, fee_limit_msat: int ) -> Payment: res = await send_payment_impl(pay_req, timeout_seconds, fee_limit_msat) - update_wallet_balance_via_sse() + _update_wallet_balance() return res -async def update_wallet_balance_via_sse(): - res = await get_wallet_balance_impl() - await send_sse_message(SSE.WALLET_BALANCE, res) - - async def get_ln_info() -> LnInfo: return await get_ln_info_impl() async def register_lightning_listener(): - await register_lightning_listener_impl() + loop = asyncio.get_event_loop() + loop.create_task(_handle_info_listener()) + loop.create_task(_handle_invoice_listener()) + + +async def _handle_info_listener(): + last_info = None + while True: + info = await get_ln_info_impl() + + if last_info != info: + status = LightningStatus.from_grpc(get_implementation_name(), info) + await send_sse_message(SSE.LN_STATUS, status.dict()) + last_info = info + + await asyncio.sleep(GATHER_INFO_INTERVALL) + + +async def _handle_invoice_listener(): + async for i in listen_invoices(): + await send_sse_message(SSE.LN_INVOICE_STATUS, i.dict()) + _update_wallet_balance() + + +def _update_wallet_balance(): + async def _perform_update(): + await asyncio.sleep(1.1) + wb = await get_wallet_balance_impl() + if _CACHE["wallet_balance"] != wb: + await send_sse_message(SSE.WALLET_BALANCE, wb.dict()) + _CACHE["wallet_balance"] = wb + + loop = asyncio.get_event_loop() + loop.create_task(_perform_update()) diff --git a/app/repositories/ln_impl/clightning.py b/app/repositories/ln_impl/clightning.py index 7478c04..a443ae7 100644 --- a/app/repositories/ln_impl/clightning.py +++ b/app/repositories/ln_impl/clightning.py @@ -25,5 +25,5 @@ async def get_ln_info_impl() -> LnInfo: raise NotImplementedError("c-lightning not yet implemented") -async def register_lightning_listener_impl(): +async def listen_invoices() -> Invoice: raise NotImplementedError("c-lightning not yet implemented") diff --git a/app/repositories/ln_impl/lnd.py b/app/repositories/ln_impl/lnd.py index 98a3bd3..11d53ab 100644 --- a/app/repositories/ln_impl/lnd.py +++ b/app/repositories/ln_impl/lnd.py @@ -17,14 +17,9 @@ from app.models.lightning import ( from app.utils import SSE from app.utils import lightning_config as lncfg from app.utils import send_sse_message -from decouple import config from fastapi.exceptions import HTTPException from starlette import status -GATHER_INFO_INTERVALL = config("gather_ln_info_interval", default=5, cast=float) - -_CACHE = {"wallet_balance": None} - def get_implementation_name() -> str: return "LND" @@ -81,7 +76,6 @@ async def send_payment_impl( async for response in lncfg.router_stub.SendPaymentV2(r): p = payment_from_grpc(response) await send_sse_message(SSE.LN_PAYMENT_STATUS, p.dict()) - _update_wallet_balance() return p except grpc.aio._call.AioRpcError as error: if ( @@ -103,52 +97,10 @@ async def get_ln_info_impl() -> LnInfo: return ln_info_from_grpc(response) -async def register_lightning_listener_impl(): - loop = asyncio.get_event_loop() - loop.create_task(_handle_invoice_listener()) - loop.create_task(_handle_get_info_gatherer_impl()) - - -async def _handle_invoice_listener(): +async def listen_invoices() -> Invoice: request = ln.InvoiceSubscription() - try: async for r in lncfg.lnd_stub.SubscribeInvoices(request): - i = invoice_from_grpc(r) - await send_sse_message(SSE.LN_INVOICE_STATUS, i.dict()) - if i.state == InvoiceState.settled: - # Wallet balance was updated, send to connected clients - _update_wallet_balance() + yield invoice_from_grpc(r) except error: print(error) - - -async def _handle_get_info_gatherer_impl(): - last_info = None - while True: - info = await get_ln_info_impl() - - if last_info != info: - await send_sse_message(SSE.LN_INFO, info.dict()) - last_info = info - - # LND doesn't provide a subscription for payment events - # This means, if the user pays a via command line or another app - # connected directly to LND, no event is fired to other clients. - # As a workaround we'll poll the wallet balance regularly and send - # it, if it's different. Related: https://github.com/lightningnetwork/lnd/pull/1962 - _update_wallet_balance() - - await asyncio.sleep(GATHER_INFO_INTERVALL) - - -def _update_wallet_balance(): - async def _perform_update(): - await asyncio.sleep(1.1) - wb = await get_wallet_balance_impl() - if _CACHE["wallet_balance"] != wb: - await send_sse_message(SSE.WALLET_BALANCE, wb.dict()) - _CACHE["wallet_balance"] = wb - - loop = asyncio.get_event_loop() - loop.create_task(_perform_update()) diff --git a/app/utils.py b/app/utils.py index deaf15a..8d9ae82 100644 --- a/app/utils.py +++ b/app/utils.py @@ -145,6 +145,7 @@ class SSE: BTC_INFO = "btc_info" LN_INFO = "ln_info" + LN_STATUS = "ln_status" LN_INVOICE_STATUS = "ln_invoice_status" LN_PAYMENT_STATUS = "ln_payment_status"