synchronous_get batch method for interface

This commit is contained in:
thomasv 2012-10-19 14:16:02 +02:00
parent 2137ab3238
commit e165c13a84
3 changed files with 37 additions and 47 deletions

View File

@ -110,7 +110,7 @@ class InterfaceAncestor(threading.Thread):
result = params[1]
params = [addr]
self.responses.put({'method':method, 'params':params, 'result':result})
self.responses.put({'method':method, 'params':params, 'result':result, 'id':msg_id})
@ -286,21 +286,22 @@ class TcpStratumInterface(InterfaceAncestor):
self.poke()
def send(self, messages):
"""return the ids of the requests that we sent"""
out = ''
ids = []
for m in messages:
method, params = m
request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
self.unanswered_requests[self.message_id] = method, params
ids.append(self.message_id)
# uncomment to debug
# print "-->",request
self.message_id += 1
out += request + '\n'
while out:
sent = self.s.send( out )
out = out[sent:]
return ids
def get_history(self, addr):
self.send([('blockchain.address.get_history', [addr])])
@ -340,9 +341,9 @@ class Interface(TcpStratumInterface, HttpStratumInterface):
def send(self, messages):
if self.protocol in 'st':
TcpStratumInterface.send(self, messages)
return TcpStratumInterface.send(self, messages)
else:
HttpStratumInterface.send(self, messages)
return HttpStratumInterface.send(self, messages)
def parse_proxy_options(self, s):
@ -378,10 +379,26 @@ class Interface(TcpStratumInterface, HttpStratumInterface):
self.is_connected = False # this exits the polling loop
self.poke()
def is_up_to_date(self):
return self.responses.empty() and not self.unanswered_requests
def synchronous_get(self, requests, timeout=100000000):
# todo: use generators, unanswered_requests should be a list of arrays...
ids = self.send(requests)
id2 = ids[:]
res = {}
while ids:
r = self.responses.get(True, timeout)
_id = r.get('id')
if _id in ids:
ids.remove(_id)
res[_id] = r.get('result')
out = []
for _id in id2:
out.append(res[_id])
return out

View File

@ -11,18 +11,7 @@ except:
i = Interface({'server':'electrum.novit.ro:50001:t'})
i.start()
i.send([('blockchain.address.get_history',[addr])])
while True:
try:
r = i.responses.get(True, 100000000000)
except KeyboardInterrupt:
break
method = r.get('method')
if method == 'blockchain.address.get_history':
confirmed = unconfirmed = 0
h = r.get('result')
h = i.synchronous_get([ ('blockchain.address.get_history',[addr]) ])[0]
for item in h:
print item['tx_hash'], item['value']
break

View File

@ -21,29 +21,6 @@ i = Interface({'server':'ecdsa.org:50002:s'})
i.start()
def get_header(i, block_height):
i.send([('blockchain.block.get_header',[block_height])])
while True:
r = i.responses.get(True, 100000000000)
method = r.get('method')
if method == 'blockchain.block.get_header':
break
return r.get('result')
def get_merkle(i, tx_hash):
i.send([('blockchain.transaction.get_merkle',[tx_hash])])
while True:
r = i.responses.get(True, 100000000000)
method = r.get('method')
if method == 'blockchain.transaction.get_merkle':
break
return r.get('result')
def merkle_root(merkle):
merkle = map (lambda tx_hash: tx_hash.decode('hex')[::-1], merkle)
@ -70,16 +47,23 @@ def hash_header(res):
def verify_tx(tx_hash):
res = get_merkle(i, tx_hash)
res = i.synchronous_get([ ('blockchain.transaction.get_merkle',[tx_hash]) ])[0]
assert tx_hash in res.get('merkle')
assert res.get('merkle_root') == merkle_root(res['merkle'])
block_height = res.get('block_height')
_hash = None
print block_height
headers_requests = []
for height in range(block_height-10,block_height+10):
header = get_header(i, height)
headers_requests.append( ('blockchain.block.get_header',[height]) )
res = i.synchronous_get(headers_requests)
_hash = None
for header in res:
if _hash: assert _hash == header.get('prev_block_hash')
_hash = hash_header(header)
#print _hash
print _hash
if height==block_height:
assert header.get('merkle_root') == res.get('merkle_root')