fix async callbacks call in jmclient code

src/jmclient/client_protocol.py

@@ -100,37 +100,42 @@ class BIP78ClientProtocol(BaseClientProtocol):
         self.defaultCallbacks(d)
 
     @commands.BIP78ReceiverUp.responder
-    def on_BIP78_RECEIVER_UP(self, hostname):
-        self.manager.bip21_uri_from_onion_hostname(hostname)
+    async def on_BIP78_RECEIVER_UP(self, hostname):
+        await self.manager.bip21_uri_from_onion_hostname(hostname)
         return {"accepted": True}
 
     @commands.BIP78ReceiverOriginalPSBT.responder
-    def on_BIP78_RECEIVER_ORIGINAL_PSBT(self, body, params):
+    async def on_BIP78_RECEIVER_ORIGINAL_PSBT(self, body, params):
         # TODO: we don't need binary key/vals client side, but will have to edit
         # PayjoinConverter for that:
-        retval = self.success_callback(body.encode("utf-8"), bdict_sdict_convert(
-            params, output_binary=True))
-        if not retval[0]:
-            d = self.callRemote(commands.BIP78ReceiverSendError, errormsg=retval[1],
-                                errorcode=retval[2])
+        cb_res = self.success_callback(
+            body.encode("utf-8"),
+            bdict_sdict_convert(params, output_binary=True))
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
+        if not cb_res[0]:
+            d = self.callRemote(commands.BIP78ReceiverSendError, errormsg=cb_res[1],
+                                errorcode=cb_res[2])
         else:
-            d = self.callRemote(commands.BIP78ReceiverSendProposal, psbt=retval[1])
+            d = self.callRemote(commands.BIP78ReceiverSendProposal, psbt=cb_res[1])
         self.defaultCallbacks(d)
         return {"accepted": True}
 
     @commands.BIP78ReceiverHiddenServiceShutdown.responder
-    def on_BIP78_RECEIVER_HIDDEN_SERVICE_SHUTDOWN(self):
+    async def on_BIP78_RECEIVER_HIDDEN_SERVICE_SHUTDOWN(self):
         """ This is called when the daemon has shut down the HS
         because of an invalid message/error. An earlier message
         will have conveyed the reason for the error.
         """
-        self.manager.shutdown()
+        await self.manager.shutdown()
         return {"accepted": True}
 
     @commands.BIP78ReceiverOnionSetupFailed.responder
-    def on_BIP78_RECEIVER_ONION_SETUP_FAILED(self, reason):
-        self.manager.info_callback(reason)
-        self.manager.shutdown()
+    async def on_BIP78_RECEIVER_ONION_SETUP_FAILED(self, reason):
+        cb_res = self.manager.info_callback(reason)
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
+        await self.manager.shutdown()
         return {"accepted": True}
 
     @commands.BIP78SenderUp.responder
@@ -149,8 +154,10 @@ class BIP78ClientProtocol(BaseClientProtocol):
         return {"accepted": True}
 
     @commands.BIP78SenderReceiveError.responder
-    def on_BIP78_SENDER_RECEIVER_ERROR(self, errormsg, errorcode):
-        self.failure_callback(errormsg, errorcode, self.manager)
+    async def on_BIP78_SENDER_RECEIVER_ERROR(self, errormsg, errorcode):
+        cb_res = self.failure_callback(errormsg, errorcode, self.manager)
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
         return {"accepted": True}
 
     @commands.BIP78InfoMsg.responder

src/jmclient/payjoin.py

@@ -1,3 +1,4 @@
+import asyncio
 from twisted.internet import reactor
 try:
     from twisted.internet.ssl import ClientContextFactory
@@ -563,7 +564,7 @@ async def send_payjoin(manager, accept_callback=None,
         reactor.connectTCP(h, p, factory)
     return (True, None)
 
-def fallback_nonpayjoin_broadcast(err, manager):
+async def fallback_nonpayjoin_broadcast(err, manager):
     """ Sends the non-coinjoin payment onto the network,
     assuming that the payjoin failed. The reason for failure is
     `err` and will usually be communicated by the server, and must
@@ -583,7 +584,9 @@ def fallback_nonpayjoin_broadcast(err, manager):
         "to see whether original payment was made.")
         log.error(errormsg)
         # ensure any GUI as well as command line sees the message:
-        manager.user_info_callback(errormsg)
+        cb_res = manager.user_info_callback(errormsg)
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
         quit()
         return
     log.info("Payment made without coinjoin. Transaction: ")
@@ -593,13 +596,13 @@ def fallback_nonpayjoin_broadcast(err, manager):
         manager.timeout_fallback_dc.cancel()
     quit()
 
-def process_error_from_server(errormsg, errorcode, manager):
+async def process_error_from_server(errormsg, errorcode, manager):
     assert isinstance(manager, JMPayjoinManager)
     # payjoin attempt has failed, we revert to standard payment.
     assert int(errorcode) != 200
     log.warn("Receiver returned error code: {}, message: {}".format(
         errorcode, errormsg))
-    fallback_nonpayjoin_broadcast(errormsg.encode("utf-8"), manager)
+    await fallback_nonpayjoin_broadcast(errormsg.encode("utf-8"), manager)
     return
 
 async def process_payjoin_proposal_from_server(response_body, manager):
@@ -609,7 +612,8 @@ async def process_payjoin_proposal_from_server(response_body, manager):
             btc.PartiallySignedTransaction.from_base64(response_body)
     except Exception as e:
         log.error("Payjoin tx from server could not be parsed: " + repr(e))
-        fallback_nonpayjoin_broadcast(b"Server sent invalid psbt", manager)
+        await fallback_nonpayjoin_broadcast(
+            b"Server sent invalid psbt", manager)
         return
     log.debug("Receiver sent us this PSBT: ")
     log.debug(manager.wallet_service.human_readable_psbt(payjoin_proposal_psbt))
@@ -626,7 +630,8 @@ async def process_payjoin_proposal_from_server(response_body, manager):
         payjoin_proposal_psbt.serialize(), with_sign_result=True)
     if err:
         log.error("Failed to sign PSBT from the receiver, error: " + err)
-        fallback_nonpayjoin_broadcast(manager, err=b"Failed to sign receiver PSBT")
+        await fallback_nonpayjoin_broadcast(
+            manager, err=b"Failed to sign receiver PSBT")
         return
 
     signresult, sender_signed_psbt = signresultandpsbt
@@ -634,7 +639,8 @@ async def process_payjoin_proposal_from_server(response_body, manager):
     success, msg = manager.set_payjoin_psbt(payjoin_proposal_psbt, sender_signed_psbt)
     if not success:
         log.error(msg)
-        fallback_nonpayjoin_broadcast(manager, err=b"Receiver PSBT checks failed.")
+        await fallback_nonpayjoin_broadcast(
+            manager, err=b"Receiver PSBT checks failed.")
         return
     # All checks have passed. We can use the already signed transaction in
     # sender_signed_psbt.
@@ -920,14 +926,20 @@ class PayjoinConverter(object):
             cb_type="unconfirmed")
         return (True, receiver_signed_psbt.to_base64(), None)
 
-    def end_receipt(self, txd, txid):
+    async def end_receipt(self, txd, txid):
         if self.manager.mode == "gui":
-            self.info_callback("Transaction seen on network, "
-                               "view wallet tab for update.:FINAL")
+            cb_res = self.info_callback("Transaction seen on network, view "
+                                        "wallet tab for update.:FINAL")
+            if asyncio.iscoroutine(cb_res):
+                cb_res = await cb_res
         else:
-            self.info_callback("Transaction seen on network: " + txid)
+            cb_res = self.info_callback("Transaction seen on network: " + txid)
+            if asyncio.iscoroutine(cb_res):
+                cb_res = await cb_res
         # in some cases (GUI) a notification of HS end is needed:
-        self.shutdown_callback()
+        cb_res = self.shutdown_callback()
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
         # informs the wallet service transaction monitor
         # that the transaction has been processed:
         return True
@@ -1017,7 +1029,7 @@ class JMBIP78ReceiverManager(object):
         else:
             return (True, a)
 
-    def bip21_uri_from_onion_hostname(self, host):
+    async def bip21_uri_from_onion_hostname(self, host):
         """ Encoding the BIP21 URI according to BIP78 specifications,
         and specifically only supporting a hidden service endpoint.
         Note: we hardcode http; no support for TLS over HS.
@@ -1032,15 +1044,21 @@ class JMBIP78ReceiverManager(object):
                                 {"amount": bip78_btc_amount,
                                  "pj": full_pj_string.encode("utf-8")},
                                 safe=":/")
-        self.info_callback("Your hidden service is available. Please\n"
-                                   "now pass this URI string to the sender to\n"
-                                   "effect the payjoin payment:")
-        self.uri_created_callback(bip21_uri)
+        cb_res = self.info_callback("Your hidden service is available. "
+                                    "Please\npass this URI string to the "
+                                    "sender to\neffect the payjoin payment:")
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
+        cb_res = self.uri_created_callback(bip21_uri)
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
         if self.mode == "command-line":
-            self.info_callback("Keep this process running until the payment "
-                               "is received.")
+            cb_res = self.info_callback("Keep this process running until the "
+                                        "payment is received.")
+            if asyncio.iscoroutine(cb_res):
+                cb_res = await cb_res
 
-    def shutdown(self):
+    async def shutdown(self):
         """ Triggered when processing has completed successfully
         or failed, receiver side.
         """
@@ -1052,6 +1070,10 @@ class JMBIP78ReceiverManager(object):
         tfdc = self.manager.timeout_fallback_dc
         if tfdc and tfdc.active():
             tfdc.cancel()
-        self.info_callback("Hidden service shutdown complete")
+        cb_res = self.info_callback("Hidden service shutdown complete")
+        if asyncio.iscoroutine(cb_res):
+            cb_res = await cb_res
         if self.shutdown_callback:
-            self.shutdown_callback()
+            cb_res = self.shutdown_callback()
+            if asyncio.iscoroutine(cb_res):
+                cb_res = await cb_res

src/jmclient/taker_utils.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import pprint
 import os
@@ -242,13 +243,17 @@ async def direct_send(wallet_service: WalletService,
         log.info(sending_info)
         if not answeryes:
             if not accept_callback:
-                if not cli_prompt_user_yesno('Would you like to push to the network?'):
-                    log.info("You chose not to broadcast the transaction, quitting.")
+                if not cli_prompt_user_yesno('Would you like to push to '
+                                             'the network?'):
+                    log.info("You chose not to broadcast the transaction, "
+                             "quitting.")
                     return False
             else:
                 accepted = accept_callback(human_readable_transaction(tx),
                                            destination, actual_amount, fee_est,
                                            custom_change_addr)
+                if asyncio.iscoroutine(accepted):
+                    accepted = await accepted
                 if not accepted:
                     return False
         if change_label:
@@ -261,13 +266,17 @@ async def direct_send(wallet_service: WalletService,
             txid = bintohex(tx.GetTxid()[::-1])
             successmsg = "Transaction sent: " + txid
             cb = log.info if not info_callback else info_callback
-            cb(successmsg)
+            cb_res = cb(successmsg)
+            if asyncio.iscoroutine(cb_res):
+                cb_res = await cb_res
             txinfo = txid if not return_transaction else tx
             return txinfo
         else:
             errormsg = "Transaction broadcast failed!"
             cb = log.error if not error_callback else error_callback
-            cb(errormsg)
+            cb_res = cb(errormsg)
+            if asyncio.iscoroutine(cb_res):
+                cb_res = await cb_res
             return False
 
 def get_tumble_log(logsdir):