#!/usr/bin/env python # # Electrum - lightweight Bitcoin client # Copyright (C) 2014 Thomas Voegtlin # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . from threading import Lock from bitcoin import Hash, hash_encode from transaction import Transaction from util import print_error, print_msg class WalletSynchronizer(): '''The synchronizer keeps the wallet up-to-date with its set of addresses and their transactions. It subscribes over the network to wallet addresses, gets the wallet to generate new addresses when necessary, requests the transaction history of any addresses we don't have the full history of, and requests binary transaction data of any transactions the wallet doesn't have. External interface: __init__() and add() member functions. ''' def __init__(self, wallet, network): self.wallet = wallet self.network = network self.new_addresses = set() # Entries are (tx_hash, tx_height) tuples self.requested_tx = set() self.requested_histories = {} self.requested_addrs = set() self.lock = Lock() self.initialize() def print_error(self, *msg): print_error("[Synchronizer]", *msg) def print_msg(self, *msg): print_msg("[Synchronizer]", *msg) def parse_response(self, response): if response.get('error'): self.print_error("response error:", response) return None, None return response['params'], response['result'] def is_up_to_date(self): return (not self.requested_tx and not self.requested_histories and not self.requested_addrs) def add(self, address): '''This can be called from the proxy or GUI threads.''' with self.lock: self.new_addresses.add(address) def subscribe_to_addresses(self, addresses): if addresses: self.requested_addrs |= addresses msgs = map(lambda addr: ('blockchain.address.subscribe', [addr]), addresses) self.network.send(msgs, self.addr_subscription_response) def addr_subscription_response(self, response): params, result = self.parse_response(response) if not params: return addr = params[0] if addr in self.requested_addrs: # Notifications won't be in self.requested_addrs.remove(addr) history = self.wallet.get_address_history(addr) if self.wallet.get_status(history) != result: if self.requested_histories.get(addr) is None: self.network.send([('blockchain.address.get_history', [addr])], self.addr_history_response) self.requested_histories[addr] = result def addr_history_response(self, response): params, result = self.parse_response(response) if not params: return addr = params[0] self.print_error("receiving history", addr, len(result)) server_status = self.requested_histories.pop(addr) # Check that txids are unique hashes = set(map(lambda item: item['tx_hash'], result)) if len(hashes) != len(result): self.print_error("error: server history has non-unique txids: %s"% addr) return # Check that the status corresponds to what was announced hist = map(lambda item: (item['tx_hash'], item['height']), result) if self.wallet.get_status(hist) != server_status: self.print_error("error: status mismatch: %s" % addr) return # Store received history self.wallet.receive_history_callback(addr, hist) # Request transactions we don't have self.request_missing_txs(hist) def tx_response(self, response): params, result = self.parse_response(response) if not params: return tx_hash, tx_height = params assert tx_hash == hash_encode(Hash(result.decode('hex'))) tx = Transaction(result) try: tx.deserialize() except Exception: self.print_msg("cannot deserialize transaction, skipping", tx_hash) return self.wallet.receive_tx_callback(tx_hash, tx, tx_height) self.requested_tx.remove((tx_hash, tx_height)) self.print_error("received tx:", tx_hash, len(tx.raw)) if not self.requested_tx: self.network.trigger_callback('updated') # Updated gets called too many times from other places as # well; if we used that signal we get the notification # three times self.network.trigger_callback("new_transaction") def request_missing_txs(self, hist): # "hist" is a list of [tx_hash, tx_height] lists missing = set() for tx_hash, tx_height in hist: if self.wallet.transactions.get(tx_hash) is None: missing.add((tx_hash, tx_height)) missing -= self.requested_tx if missing: requests = [('blockchain.transaction.get', tx) for tx in missing] self.network.send(requests, self.tx_response) self.requested_tx |= missing def initialize(self): '''Check the initial state of the wallet. Subscribe to all its addresses, and request any transactions in its address history we don't have. ''' for history in self.wallet.history.values(): # Old electrum servers returned ['*'] when all history for # the address was pruned. This no longer happens but may # remain in old wallets. if history == ['*']: continue self.request_missing_txs(history) if self.requested_tx: self.print_error("missing tx", self.requested_tx) self.subscribe_to_addresses(set(self.wallet.addresses(True))) def main_loop(self): '''Called from the network proxy thread main loop.''' # 1. Create new addresses self.wallet.synchronize() # 2. Subscribe to new addresses with self.lock: addresses = self.new_addresses self.new_addresses = set() self.subscribe_to_addresses(addresses) # 3. Detect if situation has changed up_to_date = self.is_up_to_date() if up_to_date != self.wallet.is_up_to_date(): self.wallet.set_up_to_date(up_to_date) if up_to_date: self.wallet.save_transactions() self.network.trigger_callback('updated')