Open
Description
The way xrpl.js lets you do requests and event handlers with the WebSocket client is pretty handy. It would be nice if xrpl-py's sync WebSocket client supported something similar, with .request(cmd, callback) and .on(event_type, callback) methods.
I implemented a basic form of this as follows, though it's not quite production ready (it doesn't support multiple .on() callbacks nor .off(), it doesn't enforce types, etc.):
class SmartWSClient(xrpl.clients.WebsocketClient):
def __init__(self, *args, **kwargs):
self._handlers = {}
self._pending_requests = {}
self._id = 0
super().__init__(*args, **kwargs)
def on(self, event_type, callback):
"""
Map a callback function to a type of event message from the connected
server. Only supports one callback function per event type.
"""
self._handlers[event_type] = callback
def request(self, req_dict, callback):
if "id" not in req_dict:
req_dict["id"] = f"__auto_{self._id}"
self._id += 1
# Work around https://github.com/XRPLF/xrpl-py/issues/288
req_dict["method"] = req_dict["command"]
del req_dict["command"]
req = xrpl.models.requests.request.Request.from_xrpl(req_dict)
req.validate()
self._pending_requests[req.id] = callback
self.send(req)
def handle_messages(self):
for message in self:
if message.get("type") == "response":
if message.get("status") == "success":
# Remove "status": "success", since it's confusing. We raise on errors anyway.
del message["status"]
else:
raise Exception("Unsuccessful response:", message)
msg_id = message.get("id")
if msg_id in self._pending_requests:
self._pending_requests[msg_id](message)
del self._pending_requests[msg_id]
else:
raise Exception("Response to unknown request:", message)
elif message.get("type") in self._handlers:
self._handlers[message.get("type")](message)To use this, you do something like this:
with SmartWSClient(self.ws_url) as client:
# Subscribe to ledger updates
client.request({
"command": "subscribe",
"streams": ["ledger"],
"accounts": ["rPT1Sjq2YGrBMTttX4GZHjKu9dyfzbpAYe"]
},
lambda message: print("Got subscribe response, message["result"])
)
client.on("ledgerClosed", lambda message: print("Ledger closed:", message))
client.on("transaction", lambda message: print("Got transaction notif:", message))
# Look up our balance right away
client.request({
"command": "account_info",
"account": "rPT1Sjq2YGrBMTttX4GZHjKu9dyfzbpAYe",
"ledger_index": "validated"
},
lambda m: print("Faucet balance in drops:", m["result"]["account_data"]["Balance"])
)
# Start looping through messages received. This runs indefinitely.
client.handle_messages()You can of course define more detailed handlers and pass them by name instead of using lambdas.