migrate to trio so c-lightning sockets stop hanging.

lnbits/wallets/clightning.py

@@ -3,7 +3,7 @@ try:
 except ImportError:  # pragma: nocover
     LightningRpc = None
 
-import asyncio
+import trio  # type: ignore
 import random
 import json
 
@@ -86,7 +86,7 @@ class CLightningWallet(Wallet):
         raise KeyError("supplied an invalid checking_id")
 
     async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
-        reader, writer = await asyncio.open_unix_connection(self.rpc)
+        stream = await trio.open_unix_socket(self.rpc)
 
         i = 0
         while True:
@@ -98,12 +98,9 @@ class CLightningWallet(Wallet):
                 }
             )
 
-            print(call)
-            writer.write(call.encode("ascii"))
-            await writer.drain()
+            await stream.send_all(call.encode("utf-8"))
 
-            data = await reader.readuntil(b"\n\n")
-            print(data)
+            data = await stream.receive_some()
             paid = json.loads(data.decode("ascii"))
 
             paid = self.ln.waitanyinvoice(self.last_pay_index)

lnbits/wallets/lnbits.py

@@ -1,4 +1,4 @@
-import asyncio
+import trio  # type: ignore
 from os import getenv
 from typing import Optional, Dict, AsyncGenerator
 from requests import get, post
@@ -68,5 +68,5 @@ class LNbitsWallet(Wallet):
 
     async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
         print("lnbits does not support paid invoices stream yet")
-        await asyncio.sleep(5)
+        await trio.sleep(5)
         yield ""

lnbits/wallets/lnpay.py

@@ -1,5 +1,5 @@
 import json
-import asyncio
+import trio  # type: ignore
 import httpx
 from os import getenv
 from http import HTTPStatus
@@ -77,10 +77,9 @@ class LNPayWallet(Wallet):
         return PaymentStatus(statuses[r.json()["settled"]])
 
     async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
-        self.queue: asyncio.Queue = asyncio.Queue()
-        while True:
-            yield await self.queue.get()
-            self.queue.task_done()
+        self.send, receive = trio.open_memory_channel(0)
+        async for value in receive:
+            yield value
 
     async def webhook_listener(self):
         text: str = await request.get_data()
@@ -96,6 +95,6 @@ class LNPayWallet(Wallet):
             )
             data = r.json()
             if data["settled"]:
-                self.queue.put_nowait(lntx_id)
+                self.send.send(lntx_id)
 
         return "", HTTPStatus.NO_CONTENT

lnbits/wallets/lntxbot.py

@@ -1,4 +1,4 @@
-import asyncio
+import trio  # type: ignore
 from os import getenv
 from typing import Optional, Dict, AsyncGenerator
 from requests import post
@@ -79,5 +79,5 @@ class LntxbotWallet(Wallet):
 
     async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
         print("lntxbot does not support paid invoices stream yet")
-        await asyncio.sleep(5)
+        await trio.sleep(5)
         yield ""

lnbits/wallets/opennode.py

@@ -1,5 +1,5 @@
 import json
-import asyncio
+import trio  # type: ignore
 import hmac
 import httpx
 from http import HTTPStatus
@@ -77,15 +77,12 @@ class OpenNodeWallet(Wallet):
         return PaymentStatus(statuses[r.json()["data"]["status"]])
 
     async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
-        self.queue: asyncio.Queue = asyncio.Queue()
-        while True:
-            yield await self.queue.get()
-            self.queue.task_done()
+        self.send, receive = trio.open_memory_channel(0)
+        async for value in receive:
+            yield value
 
     async def webhook_listener(self):
-        print("a request!")
         text: str = await request.get_data()
-        print("text", text)
         data = json.loads(text)
         if type(data) is not dict or "event" not in data or data["event"].get("name") != "wallet_receive":
             return "", HTTPStatus.NO_CONTENT
@@ -100,5 +97,5 @@ class OpenNodeWallet(Wallet):
             print("invalid webhook, not from opennode")
             return "", HTTPStatus.NO_CONTENT
 
-        self.queue.put_nowait(charge_id)
+        self.send.send(charge_id)
         return "", HTTPStatus.NO_CONTENT