zebra.lucky
/
joinmarket-clientserver
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

fix async callbacks call in jmclient code

zebra-lucky 2 months ago
parent
54d2ec35de
commit
4eb63be0e1
3 changed files with 80 additions and 42 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 39
      src/jmclient/client_protocol.py
  2. 66
      src/jmclient/payjoin.py
  3. 17
      src/jmclient/taker_utils.py

39
src/jmclient/client_protocol.py
Unescape View File

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

66
src/jmclient/payjoin.py
Unescape View File

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

17
src/jmclient/taker_utils.py
Unescape View File

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

Write Preview
Loading…
Cancel
Save