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
7 import thread, traceback, sys, urllib, operator
8 from json import dumps, loads
9 from Queue import Queue
12 class AbeStore(Datastore_class):
14 def __init__(self, config):
15 conf = DataStore.CONFIG_DEFAULTS
16 args, argv = readconf.parse_argv( [], conf)
17 args.dbtype = config.get('database','type')
18 if args.dbtype == 'sqlite3':
19 args.connect_args = { 'database' : config.get('database','database') }
20 elif args.dbtype == 'MySQLdb':
21 args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
22 elif args.dbtype == 'psycopg2':
23 args.connect_args = { 'database' : config.get('database','database') }
25 Datastore_class.__init__(self,args)
28 self.mempool_keys = {}
29 self.bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port'))
31 self.address_queue = Queue()
33 self.dblock = thread.allocate_lock()
37 def import_block(self, b, chain_ids=frozenset()):
39 block_id = super(AbeStore, self).import_block(b, chain_ids)
40 for pos in xrange(len(b['transactions'])):
41 tx = b['transactions'][pos]
43 tx['hash'] = util.double_sha256(tx['tx'])
44 tx_id = self.tx_find_id_and_value(tx)
46 self.update_tx_cache(tx_id)
48 print "error: import_block: no tx_id"
52 def update_tx_cache(self, txid):
53 inrows = self.get_tx_inputs(txid, False)
55 _hash = self.binout(row[6])
57 print "WARNING: missing tx_in for tx", txid
60 address = hash_to_address(chr(0), _hash)
61 if self.tx_cache.has_key(address):
62 print "cache: invalidating", address
63 self.tx_cache.pop(address)
64 self.address_queue.put(address)
66 outrows = self.get_tx_outputs(txid, False)
68 _hash = self.binout(row[6])
70 print "WARNING: missing tx_out for tx", txid
73 address = hash_to_address(chr(0), _hash)
74 if self.tx_cache.has_key(address):
75 print "cache: invalidating", address
76 self.tx_cache.pop(address)
77 self.address_queue.put(address)
79 def safe_sql(self,sql, params=(), lock=True):
81 if lock: self.dblock.acquire()
82 ret = self.selectall(sql,params)
84 print "sql error", sql
87 if lock: self.dblock.release()
92 def get_tx_outputs(self, tx_id, lock=True):
93 return self.safe_sql("""SELECT
95 txout.txout_scriptPubKey,
102 LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
103 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
104 LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
105 WHERE txout.tx_id = %d
106 ORDER BY txout.txout_pos
107 """%(tx_id), (), lock)
109 def get_tx_inputs(self, tx_id, lock=True):
110 return self.safe_sql(""" SELECT
114 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
116 COALESCE(txout.txout_pos, u.txout_pos),
119 LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
120 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
121 LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
122 LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
123 WHERE txin.tx_id = %d
124 ORDER BY txin.txin_pos
125 """%(tx_id,), (), lock)
127 def get_address_out_rows(self, dbhash):
128 return self.safe_sql(""" SELECT
138 FROM chain_candidate cc
139 JOIN block b ON (b.block_id = cc.block_id)
140 JOIN block_tx ON (block_tx.block_id = b.block_id)
141 JOIN tx ON (tx.tx_id = block_tx.tx_id)
142 JOIN txin ON (txin.tx_id = tx.tx_id)
143 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
144 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
145 WHERE pubkey.pubkey_hash = ?
146 AND cc.in_longest = 1""", (dbhash,))
148 def get_address_out_rows_memorypool(self, dbhash):
149 return self.safe_sql(""" SELECT
156 JOIN txin ON (txin.tx_id = tx.tx_id)
157 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
158 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
159 WHERE pubkey.pubkey_hash = ? """, (dbhash,))
161 def get_address_in_rows(self, dbhash):
162 return self.safe_sql(""" SELECT
172 FROM chain_candidate cc
173 JOIN block b ON (b.block_id = cc.block_id)
174 JOIN block_tx ON (block_tx.block_id = b.block_id)
175 JOIN tx ON (tx.tx_id = block_tx.tx_id)
176 JOIN txout ON (txout.tx_id = tx.tx_id)
177 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
178 WHERE pubkey.pubkey_hash = ?
179 AND cc.in_longest = 1""", (dbhash,))
181 def get_address_in_rows_memorypool(self, dbhash):
182 return self.safe_sql( """ SELECT
189 JOIN txout ON (txout.tx_id = tx.tx_id)
190 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
191 WHERE pubkey.pubkey_hash = ? """, (dbhash,))
193 def get_history(self, addr):
195 cached_version = self.tx_cache.get( addr )
196 if cached_version is not None:
197 return cached_version
199 version, binaddr = decode_check_address(addr)
203 dbhash = self.binin(binaddr)
205 rows += self.get_address_out_rows( dbhash )
206 rows += self.get_address_in_rows( dbhash )
213 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
215 print "cannot unpack row", row
217 tx_hash = self.hashout_hex(tx_hash)
219 "timestamp": int(nTime),
220 "height": int(height),
221 "is_input": int(is_in),
222 "block_hash": self.hashout_hex(blk_hash),
229 txpoints.append(txpoint)
230 known_tx.append(self.hashout_hex(tx_hash))
233 # todo: sort them really...
234 txpoints = sorted(txpoints, key=operator.itemgetter("timestamp"))
238 rows += self.get_address_in_rows_memorypool( dbhash )
239 rows += self.get_address_out_rows_memorypool( dbhash )
240 address_has_mempool = False
243 is_in, tx_hash, tx_id, pos, value = row
244 tx_hash = self.hashout_hex(tx_hash)
245 if tx_hash in known_tx:
248 # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
249 address_has_mempool = True
251 # this means pending transactions are returned by getmemorypool
252 if tx_hash not in self.mempool_keys:
255 #print "mempool", tx_hash
259 "is_input": int(is_in),
260 "block_hash": 'mempool',
266 txpoints.append(txpoint)
269 for txpoint in txpoints:
270 tx_id = txpoint['tx_id']
273 inrows = self.get_tx_inputs(tx_id)
275 _hash = self.binout(row[6])
276 address = hash_to_address(chr(0), _hash)
277 txinputs.append(address)
278 txpoint['inputs'] = txinputs
280 outrows = self.get_tx_outputs(tx_id)
282 _hash = self.binout(row[6])
283 address = hash_to_address(chr(0), _hash)
284 txoutputs.append(address)
285 txpoint['outputs'] = txoutputs
287 # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
288 if not txpoint['is_input']:
289 # detect if already redeemed...
291 if row[6] == dbhash: break
294 #row = self.get_tx_output(tx_id,dbhash)
295 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
296 # if not redeemed, we add the script
298 if not row[4]: txpoint['raw_output_script'] = row[1]
301 if not address_has_mempool:
302 self.tx_cache[addr] = txpoints
307 def get_status(self,addr):
308 # get address status, i.e. the last block for that address.
309 tx_points = self.get_history(addr)
313 lastpoint = tx_points[-1]
314 status = lastpoint['block_hash']
315 # this is a temporary hack; move it up once old clients have disappeared
316 if status == 'mempool': # and session['version'] != "old":
317 status = status + ':%d'% len(tx_points)
322 def memorypool_update(store):
324 ds = BCDataStream.BCDataStream()
325 previous_transactions = store.mempool_keys
326 store.mempool_keys = []
328 postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'})
330 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
332 if r['error'] != None:
335 v = r['result'].get('transactions')
338 ds.write(hextx.decode('hex'))
339 tx = deserialize.parse_Transaction(ds)
340 tx['hash'] = util.double_sha256(tx['tx'])
341 tx_hash = store.hashin(tx['hash'])
343 store.mempool_keys.append(tx_hash)
344 if store.tx_find_id_and_value(tx):
347 tx_id = store.import_tx(tx, False)
348 store.update_tx_cache(tx_id)
353 def send_tx(self,tx):
354 postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'})
355 respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
357 if r['error'] != None:
358 msg = r['error'].get('message')
359 out = "error: transaction rejected by memorypool: " + msg + "\n" + tx
365 def main_iteration(store):
367 store.dblock.acquire()
369 store.memorypool_update()
370 block_number = store.get_block_number(1)
373 print "IOError: cannot reach bitcoind"
376 traceback.print_exc(file=sys.stdout)
379 store.dblock.release()
385 # if there is an exception, do rollback and then re-raise the exception
386 for dircfg in store.datadirs:
388 store.catch_up_dir(dircfg)
390 store.log.exception("Failed to catch up %s", dircfg)
397 from processor import Processor
399 class BlockchainProcessor(Processor):
401 def __init__(self, config):
402 Processor.__init__(self)
403 self.store = AbeStore(config)
404 self.block_number = -1
405 self.watched_addresses = []
407 def process(self, request):
408 message_id = request['id']
409 method = request['method']
410 params = request.get('params',[])
412 if method == 'blockchain.numblocks.subscribe':
413 result = self.block_number
414 elif method == 'blockchain.address.subscribe':
416 self.watch_address(address)
417 status = self.store.get_status(address)
419 elif method == 'blockchain.address.get_history':
421 result = self.store.get_history( address )
422 elif method == 'blockchain.transaction.broadcast':
423 txo = self.store.send_tx(params[0])
424 print "sent tx:", txo
427 print "unknown method", request
430 response = { 'id':message_id, 'result':result }
431 self.push_response(response)
434 def watch_address(self, addr):
435 if addr not in self.watched_addresses:
436 self.watched_addresses.append(addr)
441 old_block_number = None
442 while not self.shared.stopped():
443 self.block_number = self.store.main_iteration()
445 if self.block_number != old_block_number:
446 old_block_number = self.block_number
447 print "block number:", self.block_number
448 self.push_response({ 'method':'blockchain.numblocks.subscribe', 'params':[self.block_number] })
452 addr = self.store.address_queue.get(False)
455 if addr in self.watched_addresses:
456 status = self.store.get_status( addr )
457 self.push_response({ 'method':'blockchain.address.subscribe', 'params':[addr, status] })