1 from Abe.abe 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
5 import psycopg2, binascii
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()
34 self.block_number = -1
35 self.watched_addresses = []
39 def import_block(self, b, chain_ids=frozenset()):
40 block_id = super(AbeStore, self).import_block(b, chain_ids)
41 for pos in xrange(len(b['transactions'])):
42 tx = b['transactions'][pos]
44 tx['hash'] = util.double_sha256(tx['tx'])
45 tx_id = self.tx_find_id_and_value(tx)
47 self.update_tx_cache(tx_id)
49 print "error: import_block: no tx_id"
53 def update_tx_cache(self, txid):
54 inrows = self.get_tx_inputs(txid, False)
56 _hash = self.binout(row[6])
57 address = hash_to_address(chr(0), _hash)
58 if self.tx_cache.has_key(address):
59 print "cache: invalidating", address
60 self.tx_cache.pop(address)
61 self.address_queue.put(address)
63 outrows = self.get_tx_outputs(txid, False)
65 _hash = self.binout(row[6])
66 address = hash_to_address(chr(0), _hash)
67 if self.tx_cache.has_key(address):
68 print "cache: invalidating", address
69 self.tx_cache.pop(address)
70 self.address_queue.put(address)
72 def safe_sql(self,sql, params=(), lock=True):
74 if lock: self.dblock.acquire()
75 ret = self.selectall(sql,params)
76 if lock: self.dblock.release()
79 print "sql error", sql
82 def get_tx_outputs(self, tx_id, lock=True):
83 return self.safe_sql("""SELECT
85 txout.txout_scriptPubKey,
92 LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
93 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
94 LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
95 WHERE txout.tx_id = %d
96 ORDER BY txout.txout_pos
97 """%(tx_id), (), lock)
99 def get_tx_inputs(self, tx_id, lock=True):
100 return self.safe_sql(""" SELECT
104 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
106 COALESCE(txout.txout_pos, u.txout_pos),
109 LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
110 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
111 LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
112 LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
113 WHERE txin.tx_id = %d
114 ORDER BY txin.txin_pos
115 """%(tx_id,), (), lock)
117 def get_address_out_rows(self, dbhash):
118 return self.safe_sql(""" SELECT
128 FROM chain_candidate cc
129 JOIN block b ON (b.block_id = cc.block_id)
130 JOIN block_tx ON (block_tx.block_id = b.block_id)
131 JOIN tx ON (tx.tx_id = block_tx.tx_id)
132 JOIN txin ON (txin.tx_id = tx.tx_id)
133 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
134 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
135 WHERE pubkey.pubkey_hash = ?
136 AND cc.in_longest = 1""", (dbhash,))
138 def get_address_out_rows_memorypool(self, dbhash):
139 return self.safe_sql(""" SELECT
146 JOIN txin ON (txin.tx_id = tx.tx_id)
147 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
148 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
149 WHERE pubkey.pubkey_hash = ? """, (dbhash,))
151 def get_address_in_rows(self, dbhash):
152 return self.safe_sql(""" SELECT
162 FROM chain_candidate cc
163 JOIN block b ON (b.block_id = cc.block_id)
164 JOIN block_tx ON (block_tx.block_id = b.block_id)
165 JOIN tx ON (tx.tx_id = block_tx.tx_id)
166 JOIN txout ON (txout.tx_id = tx.tx_id)
167 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
168 WHERE pubkey.pubkey_hash = ?
169 AND cc.in_longest = 1""", (dbhash,))
171 def get_address_in_rows_memorypool(self, dbhash):
172 return self.safe_sql( """ SELECT
179 JOIN txout ON (txout.tx_id = tx.tx_id)
180 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
181 WHERE pubkey.pubkey_hash = ? """, (dbhash,))
183 def get_history(self, addr):
185 cached_version = self.tx_cache.get( addr )
186 if cached_version is not None:
187 return cached_version
189 version, binaddr = decode_check_address(addr)
193 dbhash = self.binin(binaddr)
195 rows += self.get_address_out_rows( dbhash )
196 rows += self.get_address_in_rows( dbhash )
203 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
205 print "cannot unpack row", row
207 tx_hash = self.hashout_hex(tx_hash)
210 "height": int(height),
212 "blk_hash": self.hashout_hex(blk_hash),
219 txpoints.append(txpoint)
220 known_tx.append(self.hashout_hex(tx_hash))
223 # todo: sort them really...
224 txpoints = sorted(txpoints, key=operator.itemgetter("nTime"))
228 rows += self.get_address_in_rows_memorypool( dbhash )
229 rows += self.get_address_out_rows_memorypool( dbhash )
230 address_has_mempool = False
233 is_in, tx_hash, tx_id, pos, value = row
234 tx_hash = self.hashout_hex(tx_hash)
235 if tx_hash in known_tx:
238 # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
239 address_has_mempool = True
241 # this means pending transactions are returned by getmemorypool
242 if tx_hash not in self.mempool_keys:
245 #print "mempool", tx_hash
250 "blk_hash": 'mempool',
256 txpoints.append(txpoint)
259 for txpoint in txpoints:
260 tx_id = txpoint['tx_id']
263 inrows = self.get_tx_inputs(tx_id)
265 _hash = self.binout(row[6])
266 address = hash_to_address(chr(0), _hash)
267 txinputs.append(address)
268 txpoint['inputs'] = txinputs
270 outrows = self.get_tx_outputs(tx_id)
272 _hash = self.binout(row[6])
273 address = hash_to_address(chr(0), _hash)
274 txoutputs.append(address)
275 txpoint['outputs'] = txoutputs
277 # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
278 if not txpoint['is_in']:
279 # detect if already redeemed...
281 if row[6] == dbhash: break
284 #row = self.get_tx_output(tx_id,dbhash)
285 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
286 # if not redeemed, we add the script
288 if not row[4]: txpoint['raw_scriptPubKey'] = row[1]
291 if not address_has_mempool:
292 self.tx_cache[addr] = txpoints
297 def get_status(self,addr):
298 # get address status, i.e. the last block for that address.
299 tx_points = self.get_history(addr)
303 lastpoint = tx_points[-1]
304 status = lastpoint['blk_hash']
305 # this is a temporary hack; move it up once old clients have disappeared
306 if status == 'mempool': # and session['version'] != "old":
307 status = status + ':%d'% len(tx_points)
312 def memorypool_update(store):
314 ds = BCDataStream.BCDataStream()
315 previous_transactions = store.mempool_keys
316 store.mempool_keys = []
318 postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'})
320 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
322 if r['error'] != None:
325 v = r['result'].get('transactions')
328 ds.write(hextx.decode('hex'))
329 tx = deserialize.parse_Transaction(ds)
330 tx['hash'] = util.double_sha256(tx['tx'])
331 tx_hash = store.hashin(tx['hash'])
333 store.mempool_keys.append(tx_hash)
334 if store.tx_find_id_and_value(tx):
337 tx_id = store.import_tx(tx, False)
338 store.update_tx_cache(tx_id)
343 def send_tx(self,tx):
344 postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'})
345 respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
347 if r['error'] != None:
348 out = "error: transaction rejected by memorypool\n"+tx
354 def main_iteration(store):
356 store.dblock.acquire()
358 store.memorypool_update()
359 block_number = store.get_block_number(1)
362 print "IOError: cannot reach bitcoind"
365 traceback.print_exc(file=sys.stdout)
368 store.dblock.release()
372 def watch_address(self, addr):
373 if addr not in self.watched_addresses:
374 self.watched_addresses.append(addr)
376 def run(self, processor):
378 old_block_number = None
379 while not processor.shared.stopped():
380 self.block_number = self.main_iteration()
382 if self.block_number != old_block_number:
383 old_block_number = self.block_number
384 processor.push_response({ 'method':'numblocks.subscribe', 'result':self.block_number })
388 addr = self.address_queue.get(False)
391 if addr in self.watched_addresses:
392 status = self.get_status( addr )
393 processor.push_response({ 'method':'address.subscribe', 'params':[addr], 'result':status })