WebSocket Sync client: support attaching event handlers, making a request with callback

[mDuo13]

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.

[mvadari]

I have a slightly different implementation here: https://github.com/XRPLF/xrpl-py/blob/sidechain/xrpl/asyncio/clients/websocket_base.py#L137 It's on a separate branch for now, but it'll likely be merged in the next month or so.

[mvadari]

For posterity, here's the commit: https://github.com/XRPLF/xrpl-py/commit/677a94be663e21347aa841ef1662a58d34686e2a

I don't plan on adding this feature, as it would need to be well-tested. I implemented this as a part of a previous sidechain design; it is no longer necessary.