tImplement scripthash subscriptions. - obelisk - Electrum server using libbitcoin as its backend
HTML git clone https://git.parazyd.org/obelisk
DIR Log
DIR Files
DIR Refs
DIR README
DIR LICENSE
---
DIR commit 25e8f42479f13b252cb2923e44919dcac38f2950
DIR parent 95697b92c9c29b27f006815b62dc8eb27522e2b5
HTML Author: parazyd <parazyd@dyne.org>
Date: Fri, 9 Apr 2021 01:23:20 +0200
Implement scripthash subscriptions.
In theory this works, but in practice and v4 - it doesn't.
Needs more investigation.
Diffstat:
M electrumobelisk/protocol.py | 33 +++++++++++++++++++++++++------
M electrumobelisk/zeromq.py | 16 ++++++++++++++++
2 files changed, 43 insertions(+), 6 deletions(-)
---
DIR diff --git a/electrumobelisk/protocol.py b/electrumobelisk/protocol.py
t@@ -63,7 +63,6 @@ class ElectrumProtocol(asyncio.Protocol): # pylint: disable=R0904,R0902
# Consider renaming bx to something else
self.bx = Client(log, endpoints, self.loop)
self.block_queue = None
- self.tx_queue = None
# TODO: Clean up on client disconnect
self.tasks = []
self.sh_subscriptions = {}
t@@ -115,6 +114,9 @@ class ElectrumProtocol(asyncio.Protocol): # pylint: disable=R0904,R0902
"""Destructor function"""
self.log.debug("ElectrumProtocol.stop()")
if self.bx:
+ for i in self.sh_subscriptions:
+ self.log.debug("bx.unsubscribe %s", i)
+ await self.bx.unsubscribe_scripthash(i)
await self.bx.stop()
self.stopped = True
t@@ -352,6 +354,18 @@ class ElectrumProtocol(asyncio.Protocol): # pylint: disable=R0904,R0902
})
return {"result": ret}
+ async def scripthash_notifier(self, writer, scripthash):
+ # TODO: Figure out how this actually works
+ _ec, sh_queue = await self.bx.subscribe_scripthash(scripthash)
+ if _ec and ec != 0:
+ self.log.error("bx.subscribe_scripthash failed:", repr(_ec))
+ return
+
+ while True:
+ # item = (seq, height, block_data)
+ item = await sh_queue.get()
+ self.log.debug("sh_subscription item: %s", item)
+
async def blockchain_scripthash_subscribe(self, writer, query): # pylint: disable=W0613
"""Method: blockchain.scripthash.subscribe
Subscribe to a script hash.
t@@ -367,8 +381,9 @@ class ElectrumProtocol(asyncio.Protocol): # pylint: disable=R0904,R0902
if _ec and _ec != 0:
return {"error": "request corrupted"}
- # TODO: task for tx subscription
- self.sh_subscriptions[scripthash] = "foo"
+ task = asyncio.create_task(self.scripthash_notifier(
+ writer, scripthash))
+ self.sh_subscriptions[scripthash] = {"task": task}
if len(history) < 1:
return {"result": None}
t@@ -380,10 +395,13 @@ class ElectrumProtocol(asyncio.Protocol): # pylint: disable=R0904,R0902
status.append(safe_hexlify(i[kind]["hash"][::-1]))
status.append(str(i[kind]["height"])) # str because of join
+ self.sh_subscriptions[scripthash]["status"] = status
+ return {"result": self.scripthash_status(status)}
+
+ def scripthash_status(self, status):
# TODO: Check if trailing colon is necessary
concat = ":".join(status) + ":"
- res = hash_to_hex_str(sha256(concat.encode()))
- return {"result": res}
+ return hash_to_hex_str(sha256(concat.encode()))
async def blockchain_scripthash_unsubscribe(self, writer, query): # pylint: disable=W0613
"""Method: blockchain.scripthash.unsubscribe
t@@ -398,8 +416,11 @@ class ElectrumProtocol(asyncio.Protocol): # pylint: disable=R0904,R0902
return {"error": "invalid scripthash"}
if scripthash in self.sh_subscriptions:
- self.sh_subscriptions[scripthash].cancel()
+ self.sh_subscriptions[scripthash]["task"].cancel()
+ await self.bx.unsubscribe_scripthash(scripthash)
+ del self.sh_subscriptions[scripthash]
return {"result": True}
+
return {"result": False}
async def blockchain_transaction_broadcast(self, writer, query): # pylint: disable=W0613
DIR diff --git a/electrumobelisk/zeromq.py b/electrumobelisk/zeromq.py
t@@ -337,6 +337,22 @@ class Client:
return error_code, None
return error_code, bh2u(data)
+ async def subscribe_scripthash(self, scripthash):
+ """Subscribe to scripthash"""
+ command = b"subscribe.key"
+ decoded_address = unhexlify(scripthash)
+ return await self._subscription_request(command, decoded_address)
+
+ async def unsubscribe_scripthash(self, scripthash):
+ """Unsubscribe scripthash"""
+ # TODO: This call should ideally also remove the subscription
+ # request from the RequestCollection.
+ # This call solicits a final call from the server with an
+ # `error::service_stopped` error code.
+ command = b"unsubscribe.key"
+ decoded_address = unhexlify(scripthash)
+ return await self._simple_request(command, decoded_address)
+
async def fetch_history4(self, scripthash, height=0):
"""Fetch history for given scripthash"""
# BUG: There is something strange happening sometimes, for example