backends/abe/__init__.py

   1 from Abe.util import hash_to_address, decode_check_address
   2 from Abe.DataStore import DataStore as Datastore_class
   3 from Abe import DataStore, readconf, BCDataStream,  deserialize, util, base58
   4
   5 import binascii
   6
   7 import thread, traceback, sys, urllib, operator
   8 from json import dumps, loads
   9 from Queue import Queue
  10 import time, threading
  11
  12
  13 import hashlib
  14 encode = lambda x: x[::-1].encode('hex')
  15 decode = lambda x: x.decode('hex')[::-1]
  16 Hash = lambda x: hashlib.sha256(hashlib.sha256(x).digest()).digest()
  17
  18 def rev_hex(s):
  19     return s.decode('hex')[::-1].encode('hex')
  20
  21 def int_to_hex(i, length=1):
  22     s = hex(i)[2:].rstrip('L')
  23     s = "0"*(2*length - len(s)) + s
  24     return rev_hex(s)
  25
  26 def header_to_string(res):
  27     s = int_to_hex(res.get('version'),4) \
  28         + rev_hex(res.get('prev_block_hash')) \
  29         + rev_hex(res.get('merkle_root')) \
  30         + int_to_hex(int(res.get('timestamp')),4) \
  31         + int_to_hex(int(res.get('bits')),4) \
  32         + int_to_hex(int(res.get('nonce')),4)
  33     return s
  34
  35
  36 class AbeStore(Datastore_class):
  37
  38     def __init__(self, config):
  39         conf = DataStore.CONFIG_DEFAULTS
  40         args, argv = readconf.parse_argv( [], conf)
  41         args.dbtype = config.get('database','type')
  42         if args.dbtype == 'sqlite3':
  43             args.connect_args = { 'database' : config.get('database','database') }
  44         elif args.dbtype == 'MySQLdb':
  45             args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
  46         elif args.dbtype == 'psycopg2':
  47             args.connect_args = { 'database' : config.get('database','database') }
  48
  49         coin = config.get('server', 'coin')
  50         self.addrtype = 0
  51         if coin == 'litecoin':
  52             print 'Litecoin settings:'
  53             datadir = config.get('server','datadir')
  54             print '  datadir = ' + datadir
  55             args.datadir = [{"dirname":datadir,"chain":"Litecoin","code3":"LTC","address_version":"\u0030"}]
  56             print '  addrtype = 48'
  57             self.addrtype = 48
  58
  59         Datastore_class.__init__(self,args)
  60
  61         # Use 1 (Bitcoin) if chain_id is not sent
  62         self.chain_id = self.datadirs[0]["chain_id"] or 1
  63         print 'Coin chain_id = %d' % self.chain_id
  64
  65         self.sql_limit = int( config.get('database','limit') )
  66
  67         self.tx_cache = {}
  68         self.bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port'))
  69
  70         self.address_queue = Queue()
  71
  72         self.dblock = thread.allocate_lock()
  73         self.last_tx_id = 0
  74         self.known_mempool_hashes = []
  75
  76
  77     def import_tx(self, tx, is_coinbase):
  78         tx_id = super(AbeStore, self).import_tx(tx, is_coinbase)
  79         self.last_tx_id = tx_id
  80         return tx_id
  81
  82
  83
  84
  85     def import_block(self, b, chain_ids=frozenset()):
  86         #print "import block"
  87         block_id = super(AbeStore, self).import_block(b, chain_ids)
  88         for pos in xrange(len(b['transactions'])):
  89             tx = b['transactions'][pos]
  90             if 'hash' not in tx:
  91                 tx['hash'] = util.double_sha256(tx['tx'])
  92             tx_id = self.tx_find_id_and_value(tx)
  93             if tx_id:
  94                 self.update_tx_cache(tx_id)
  95             else:
  96                 print "error: import_block: no tx_id"
  97         return block_id
  98
  99
 100     def update_tx_cache(self, txid):
 101         inrows = self.get_tx_inputs(txid, False)
 102         for row in inrows:
 103             _hash = self.binout(row[6])
 104             if not _hash:
 105                 #print "WARNING: missing tx_in for tx", txid
 106                 continue
 107
 108             address = hash_to_address(chr(self.addrtype), _hash)
 109             if self.tx_cache.has_key(address):
 110                 print "cache: invalidating", address
 111                 self.tx_cache.pop(address)
 112             self.address_queue.put(address)
 113
 114         outrows = self.get_tx_outputs(txid, False)
 115         for row in outrows:
 116             _hash = self.binout(row[6])
 117             if not _hash:
 118                 #print "WARNING: missing tx_out for tx", txid
 119                 continue
 120
 121             address = hash_to_address(chr(self.addrtype), _hash)
 122             if self.tx_cache.has_key(address):
 123                 print "cache: invalidating", address
 124                 self.tx_cache.pop(address)
 125             self.address_queue.put(address)
 126
 127     def safe_sql(self,sql, params=(), lock=True):
 128
 129         error = False
 130         try:
 131             if lock: self.dblock.acquire()
 132             ret = self.selectall(sql,params)
 133         except:
 134             error = True
 135             traceback.print_exc(file=sys.stdout)
 136         finally:
 137             if lock: self.dblock.release()
 138
 139         if error:
 140             raise BaseException('sql error')
 141
 142         return ret
 143
 144
 145     def get_tx_outputs(self, tx_id, lock=True):
 146         return self.safe_sql("""SELECT
 147                 txout.txout_pos,
 148                 txout.txout_scriptPubKey,
 149                 txout.txout_value,
 150                 nexttx.tx_hash,
 151                 nexttx.tx_id,
 152                 txin.txin_pos,
 153                 pubkey.pubkey_hash
 154               FROM txout
 155               LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
 156               LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
 157               LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
 158              WHERE txout.tx_id = %d
 159              ORDER BY txout.txout_pos
 160         """%(tx_id), (), lock)
 161
 162     def get_tx_inputs(self, tx_id, lock=True):
 163         return self.safe_sql(""" SELECT
 164                 txin.txin_pos,
 165                 txin.txin_scriptSig,
 166                 txout.txout_value,
 167                 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
 168                 prevtx.tx_id,
 169                 COALESCE(txout.txout_pos, u.txout_pos),
 170                 pubkey.pubkey_hash
 171               FROM txin
 172               LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
 173               LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
 174               LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
 175               LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
 176              WHERE txin.tx_id = %d
 177              ORDER BY txin.txin_pos
 178              """%(tx_id,), (), lock)
 179
 180
 181     def get_address_out_rows(self, dbhash):
 182         out = self.safe_sql(""" SELECT
 183                 b.block_nTime,
 184                 cc.chain_id,
 185                 b.block_height,
 186                 1,
 187                 b.block_hash,
 188                 tx.tx_hash,
 189                 tx.tx_id,
 190                 txin.txin_pos,
 191                 -prevout.txout_value
 192               FROM chain_candidate cc
 193               JOIN block b ON (b.block_id = cc.block_id)
 194               JOIN block_tx ON (block_tx.block_id = b.block_id)
 195               JOIN tx ON (tx.tx_id = block_tx.tx_id)
 196               JOIN txin ON (txin.tx_id = tx.tx_id)
 197               JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
 198               JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
 199              WHERE pubkey.pubkey_hash = ?
 200                AND cc.chain_id = ?
 201                AND cc.in_longest = 1
 202              LIMIT ? """, (dbhash, self.chain_id, self.sql_limit))
 203
 204         if len(out)==self.sql_limit:
 205             raise BaseException('limit reached')
 206         return out
 207
 208     def get_address_out_rows_memorypool(self, dbhash):
 209         out = self.safe_sql(""" SELECT
 210                 1,
 211                 tx.tx_hash,
 212                 tx.tx_id,
 213                 txin.txin_pos,
 214                 -prevout.txout_value
 215               FROM tx
 216               JOIN txin ON (txin.tx_id = tx.tx_id)
 217               JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
 218               JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
 219              WHERE pubkey.pubkey_hash = ?
 220              LIMIT ? """, (dbhash,self.sql_limit))
 221
 222         if len(out)==self.sql_limit:
 223             raise BaseException('limit reached')
 224         return out
 225
 226     def get_address_in_rows(self, dbhash):
 227         out = self.safe_sql(""" SELECT
 228                 b.block_nTime,
 229                 cc.chain_id,
 230                 b.block_height,
 231                 0,
 232                 b.block_hash,
 233                 tx.tx_hash,
 234                 tx.tx_id,
 235                 txout.txout_pos,
 236                 txout.txout_value
 237               FROM chain_candidate cc
 238               JOIN block b ON (b.block_id = cc.block_id)
 239               JOIN block_tx ON (block_tx.block_id = b.block_id)
 240               JOIN tx ON (tx.tx_id = block_tx.tx_id)
 241               JOIN txout ON (txout.tx_id = tx.tx_id)
 242               JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
 243              WHERE pubkey.pubkey_hash = ?
 244                AND cc.chain_id = ?
 245                AND cc.in_longest = 1
 246                LIMIT ? """, (dbhash, self.chain_id, self.sql_limit))
 247
 248         if len(out)==self.sql_limit:
 249             raise BaseException('limit reached')
 250         return out
 251
 252     def get_address_in_rows_memorypool(self, dbhash):
 253         out = self.safe_sql( """ SELECT
 254                 0,
 255                 tx.tx_hash,
 256                 tx.tx_id,
 257                 txout.txout_pos,
 258                 txout.txout_value
 259               FROM tx
 260               JOIN txout ON (txout.tx_id = tx.tx_id)
 261               JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
 262              WHERE pubkey.pubkey_hash = ?
 263              LIMIT ? """, (dbhash,self.sql_limit))
 264
 265         if len(out)==self.sql_limit:
 266             raise BaseException('limit reached')
 267         return out
 268
 269     def get_history(self, addr):
 270
 271         cached_version = self.tx_cache.get( addr )
 272         if cached_version is not None:
 273             return cached_version
 274
 275         version, binaddr = decode_check_address(addr)
 276         if binaddr is None:
 277             return None
 278
 279         dbhash = self.binin(binaddr)
 280         rows = []
 281         rows += self.get_address_out_rows( dbhash )
 282         rows += self.get_address_in_rows( dbhash )
 283
 284         txpoints = []
 285         known_tx = []
 286
 287         for row in rows:
 288             try:
 289                 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
 290             except:
 291                 print "cannot unpack row", row
 292                 break
 293             tx_hash = self.hashout_hex(tx_hash)
 294             txpoint = {
 295                     "timestamp":    int(nTime),
 296                     "height":   int(height),
 297                     "is_input":    int(is_in),
 298                     "block_hash": self.hashout_hex(blk_hash),
 299                     "tx_hash":  tx_hash,
 300                     "tx_id":    int(tx_id),
 301                     "index":      int(pos),
 302                     "value":    int(value),
 303                     }
 304
 305             txpoints.append(txpoint)
 306             known_tx.append(self.hashout_hex(tx_hash))
 307
 308
 309         # todo: sort them really...
 310         txpoints = sorted(txpoints, key=operator.itemgetter("timestamp"))
 311
 312         # read memory pool
 313         rows = []
 314         rows += self.get_address_in_rows_memorypool( dbhash )
 315         rows += self.get_address_out_rows_memorypool( dbhash )
 316         address_has_mempool = False
 317
 318         for row in rows:
 319             is_in, tx_hash, tx_id, pos, value = row
 320             tx_hash = self.hashout_hex(tx_hash)
 321             if tx_hash in known_tx:
 322                 continue
 323
 324             # discard transactions that are too old
 325             if self.last_tx_id - tx_id > 50000:
 326                 print "discarding tx id", tx_id
 327                 continue
 328
 329             # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
 330             address_has_mempool = True
 331
 332             #print "mempool", tx_hash
 333             txpoint = {
 334                     "timestamp":    0,
 335                     "height":   0,
 336                     "is_input":    int(is_in),
 337                     "block_hash": 'mempool',
 338                     "tx_hash":  tx_hash,
 339                     "tx_id":    int(tx_id),
 340                     "index":      int(pos),
 341                     "value":    int(value),
 342                     }
 343             txpoints.append(txpoint)
 344
 345
 346         for txpoint in txpoints:
 347             tx_id = txpoint['tx_id']
 348
 349             txinputs = []
 350             inrows = self.get_tx_inputs(tx_id)
 351             for row in inrows:
 352                 _hash = self.binout(row[6])
 353                 if not _hash:
 354                     #print "WARNING: missing tx_in for tx", tx_id, addr
 355                     continue
 356                 address = hash_to_address(chr(self.addrtype), _hash)
 357                 txinputs.append(address)
 358             txpoint['inputs'] = txinputs
 359             txoutputs = []
 360             outrows = self.get_tx_outputs(tx_id)
 361             for row in outrows:
 362                 _hash = self.binout(row[6])
 363                 if not _hash:
 364                     #print "WARNING: missing tx_out for tx", tx_id, addr
 365                     continue
 366                 address = hash_to_address(chr(self.addrtype), _hash)
 367                 txoutputs.append(address)
 368             txpoint['outputs'] = txoutputs
 369
 370             # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
 371             if not txpoint['is_input']:
 372                 # detect if already redeemed...
 373                 for row in outrows:
 374                     if row[6] == dbhash: break
 375                 else:
 376                     raise
 377                 #row = self.get_tx_output(tx_id,dbhash)
 378                 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
 379                 # if not redeemed, we add the script
 380                 if row:
 381                     if not row[4]: txpoint['raw_output_script'] = row[1]
 382
 383             txpoint.pop('tx_id')
 384
 385         # cache result
 386         # do not cache mempool results because statuses are ambiguous
 387         if not address_has_mempool:
 388             self.tx_cache[addr] = txpoints
 389
 390         return txpoints
 391
 392
 393     def get_status(self,addr):
 394         # get address status, i.e. the last block for that address.
 395         tx_points = self.get_history(addr)
 396         if not tx_points:
 397             status = None
 398         else:
 399             lastpoint = tx_points[-1]
 400             status = lastpoint['block_hash']
 401             # this is a temporary hack; move it up once old clients have disappeared
 402             if status == 'mempool': # and session['version'] != "old":
 403                 status = status + ':%d'% len(tx_points)
 404         return status
 405
 406
 407     def get_block_header(self, block_height):
 408         out = self.safe_sql("""
 409             SELECT
 410                 block_hash,
 411                 block_version,
 412                 block_hashMerkleRoot,
 413                 block_nTime,
 414                 block_nBits,
 415                 block_nNonce,
 416                 block_height,
 417                 prev_block_hash,
 418                 block_id
 419               FROM chain_summary
 420              WHERE block_height = %d AND in_longest = 1"""%block_height)
 421
 422         if not out: raise BaseException("block not found")
 423         row = out[0]
 424         (block_hash, block_version, hashMerkleRoot, nTime, nBits, nNonce, height,prev_block_hash, block_id) \
 425             = ( self.hashout_hex(row[0]), int(row[1]), self.hashout_hex(row[2]), int(row[3]), int(row[4]), int(row[5]), int(row[6]), self.hashout_hex(row[7]), int(row[8]) )
 426
 427         out = {"block_height":block_height, "version":block_version, "prev_block_hash":prev_block_hash,
 428                 "merkle_root":hashMerkleRoot, "timestamp":nTime, "bits":nBits, "nonce":nNonce}
 429         return out
 430
 431
 432     def get_chunk(self, index):
 433         sql = """
 434             SELECT
 435                 block_hash,
 436                 block_version,
 437                 block_hashMerkleRoot,
 438                 block_nTime,
 439                 block_nBits,
 440                 block_nNonce,
 441                 block_height,
 442                 prev_block_hash,
 443                 block_height
 444               FROM chain_summary
 445              WHERE block_height >= %d AND block_height< %d AND in_longest = 1"""%(index*2016, (index+1)*2016)
 446
 447         out = self.safe_sql(sql)
 448         msg = ''
 449         for row in out:
 450             (block_hash, block_version, hashMerkleRoot, nTime, nBits, nNonce, height, prev_block_hash, block_height) \
 451                 = ( self.hashout_hex(row[0]), int(row[1]), self.hashout_hex(row[2]), int(row[3]), int(row[4]), int(row[5]), int(row[6]), self.hashout_hex(row[7]), int(row[8]) )
 452             h = {"block_height":block_height, "version":block_version, "prev_block_hash":prev_block_hash,
 453                    "merkle_root":hashMerkleRoot, "timestamp":nTime, "bits":nBits, "nonce":nNonce}
 454
 455             if h.get('block_height')==0: h['prev_block_hash'] = "0"*64
 456             msg += header_to_string(h)
 457
 458             #print "hash", encode(Hash(msg.decode('hex')))
 459             #if h.get('block_height')==1:break
 460
 461         print "get_chunk", index, len(msg)
 462         return msg
 463
 464
 465
 466     def get_tx_merkle(self, tx_hash):
 467
 468         out = self.safe_sql("""
 469              SELECT block_tx.block_id FROM tx
 470              JOIN block_tx on tx.tx_id = block_tx.tx_id
 471              JOIN chain_summary on chain_summary.block_id = block_tx.block_id
 472              WHERE tx_hash='%s' AND in_longest = 1"""%tx_hash)
 473         block_id = out[0]
 474
 475         # get block height
 476         out = self.safe_sql("SELECT block_height FROM chain_summary WHERE block_id = %d AND in_longest = 1"%block_id)
 477
 478         if not out: raise BaseException("block not found")
 479         block_height = int(out[0][0])
 480
 481         merkle = []
 482         tx_pos = None
 483
 484         # list all tx in block
 485         for row in self.safe_sql("""
 486             SELECT DISTINCT tx_id, tx_pos, tx_hash
 487               FROM txin_detail
 488              WHERE block_id = ?
 489              ORDER BY tx_pos""", (block_id,)):
 490             _id, _pos, _hash = row
 491             merkle.append(_hash)
 492             if _hash == tx_hash: tx_pos = int(_pos)
 493
 494         # find subset.
 495         # TODO: do not compute this on client request, better store the hash tree of each block in a database...
 496
 497         merkle = map(decode, merkle)
 498         target_hash = decode(tx_hash)
 499
 500         s = []
 501         while len(merkle) != 1:
 502             if len(merkle)%2: merkle.append( merkle[-1] )
 503             n = []
 504             while merkle:
 505                 new_hash = Hash( merkle[0] + merkle[1] )
 506                 if merkle[0] == target_hash:
 507                     s.append( encode(merkle[1]))
 508                     target_hash = new_hash
 509                 elif merkle[1] == target_hash:
 510                     s.append( encode(merkle[0]))
 511                     target_hash = new_hash
 512                 n.append( new_hash )
 513                 merkle = merkle[2:]
 514             merkle = n
 515
 516         # send result
 517         return {"block_height":block_height, "merkle":s, "pos":tx_pos}
 518
 519
 520
 521
 522     def memorypool_update(store):
 523
 524         ds = BCDataStream.BCDataStream()
 525         postdata = dumps({"method": 'getrawmempool', 'params': [], 'id':'jsonrpc'})
 526         respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
 527         r = loads(respdata)
 528         if r['error'] != None:
 529             print r['error']
 530             return
 531
 532         mempool_hashes = r.get('result')
 533         for tx_hash in mempool_hashes:
 534
 535             if tx_hash in store.known_mempool_hashes: continue
 536             store.known_mempool_hashes.append(tx_hash)
 537
 538             postdata = dumps({"method": 'getrawtransaction', 'params': [tx_hash], 'id':'jsonrpc'})
 539             respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
 540             r = loads(respdata)
 541             if r['error'] != None:
 542                 continue
 543             hextx = r.get('result')
 544             ds.clear()
 545             ds.write(hextx.decode('hex'))
 546             tx = deserialize.parse_Transaction(ds)
 547             tx['hash'] = util.double_sha256(tx['tx'])
 548
 549             if store.tx_find_id_and_value(tx):
 550                 pass
 551             else:
 552                 tx_id = store.import_tx(tx, False)
 553                 store.update_tx_cache(tx_id)
 554                 #print tx_hash
 555
 556         store.commit()
 557         store.known_mempool_hashes = mempool_hashes
 558
 559
 560     def send_tx(self,tx):
 561         postdata = dumps({"method": 'sendrawtransaction', 'params': [tx], 'id':'jsonrpc'})
 562         respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
 563         r = loads(respdata)
 564         if r['error'] != None:
 565             msg = r['error'].get('message')
 566             out = "error: transaction rejected by memorypool: " + msg + "\n" + tx
 567         else:
 568             out = r['result']
 569         return out
 570
 571
 572     def main_iteration(store):
 573         with store.dblock:
 574             store.catch_up()
 575             store.memorypool_update()
 576             height = store.get_block_number( store.chain_id )
 577
 578         block_header = store.get_block_header( height )
 579         return block_header
 580
 581
 582
 583
 584     def catch_up(store):
 585         # if there is an exception, do rollback and then re-raise the exception
 586         for dircfg in store.datadirs:
 587             try:
 588                 store.catch_up_dir(dircfg)
 589             except Exception, e:
 590                 store.log.exception("Failed to catch up %s", dircfg)
 591                 store.rollback()
 592                 raise e
 593
 594
 595
 596
 597 from processor import Processor
 598
 599 class BlockchainProcessor(Processor):
 600
 601     def __init__(self, config):
 602         Processor.__init__(self)
 603         self.store = AbeStore(config)
 604         self.watched_addresses = []
 605
 606         # catch_up first
 607         self.block_header = self.store.main_iteration()
 608         self.block_number = self.block_header.get('block_height')
 609         print "blockchain: %d blocks"%self.block_number
 610
 611         threading.Timer(10, self.run_store_iteration).start()
 612
 613     def process(self, request):
 614         #print "abe process", request
 615
 616         message_id = request['id']
 617         method = request['method']
 618         params = request.get('params',[])
 619         result = None
 620         error = None
 621
 622         if method == 'blockchain.numblocks.subscribe':
 623             result = self.block_number
 624
 625         elif method == 'blockchain.headers.subscribe':
 626             result = self.block_header
 627
 628         elif method == 'blockchain.address.subscribe':
 629             try:
 630                 address = params[0]
 631                 result = self.store.get_status(address)
 632                 self.watch_address(address)
 633             except BaseException, e:
 634                 error = str(e) + ': ' + address
 635                 print "error:", error
 636
 637         elif method == 'blockchain.address.get_history':
 638             try:
 639                 address = params[0]
 640                 result = self.store.get_history( address )
 641             except BaseException, e:
 642                 error = str(e) + ': ' + address
 643                 print "error:", error
 644
 645         elif method == 'blockchain.block.get_header':
 646             try:
 647                 height = params[0]
 648                 result = self.store.get_block_header( height )
 649             except BaseException, e:
 650                 error = str(e) + ': %d'% height
 651                 print "error:", error
 652
 653         elif method == 'blockchain.block.get_chunk':
 654             try:
 655                 index = params[0]
 656                 result = self.store.get_chunk( index )
 657             except BaseException, e:
 658                 error = str(e) + ': %d'% index
 659                 print "error:", error
 660
 661         elif method == 'blockchain.transaction.broadcast':
 662             txo = self.store.send_tx(params[0])
 663             print "sent tx:", txo
 664             result = txo
 665
 666         elif method == 'blockchain.transaction.get_merkle':
 667             try:
 668                 tx_hash = params[0]
 669                 result = self.store.get_tx_merkle(tx_hash )
 670             except BaseException, e:
 671                 error = str(e) + ': ' + tx_hash
 672                 print "error:", error
 673
 674         else:
 675             error = "unknown method:%s"%method
 676
 677
 678         if error:
 679             response = { 'id':message_id, 'error':error }
 680             self.push_response(response)
 681         elif result != '':
 682             response = { 'id':message_id, 'result':result }
 683             self.push_response(response)
 684
 685
 686     def watch_address(self, addr):
 687         if addr not in self.watched_addresses:
 688             self.watched_addresses.append(addr)
 689
 690
 691     def run_store_iteration(self):
 692
 693         try:
 694             block_header = self.store.main_iteration()
 695         except:
 696             traceback.print_exc(file=sys.stdout)
 697             print "terminating"
 698             self.shared.stop()
 699
 700         if self.shared.stopped():
 701             print "exit timer"
 702             return
 703
 704         if self.block_number != block_header.get('block_height'):
 705             self.block_number = block_header.get('block_height')
 706             print "block number:", self.block_number
 707             self.push_response({ 'id': None, 'method':'blockchain.numblocks.subscribe', 'params':[self.block_number] })
 708
 709         if self.block_header != block_header:
 710             self.block_header = block_header
 711             self.push_response({ 'id': None, 'method':'blockchain.headers.subscribe', 'params':[self.block_header] })
 712
 713         while True:
 714             try:
 715                 addr = self.store.address_queue.get(False)
 716             except:
 717                 break
 718             if addr in self.watched_addresses:
 719                 status = self.store.get_status( addr )
 720                 self.push_response({ 'id': None, 'method':'blockchain.address.subscribe', 'params':[addr, status] })
 721
 722         threading.Timer(10, self.run_store_iteration).start()
 723
 724