import random
import aes
import ecdsa
+import Queue
+import time
from ecdsa.util import string_to_number, number_to_string
from util import print_error, user_dir, format_satoshis
self.config = config
self.electrum_version = ELECTRUM_VERSION
- self.update_callbacks = []
# saved fields
self.seed_version = config.get('seed_version', SEED_VERSION)
if self.seed_version != SEED_VERSION:
raise ValueError("This wallet seed is deprecated. Please run upgrade.py for a diagnostic.")
-
- def register_callback(self, update_callback):
- with self.lock:
- self.update_callbacks.append(update_callback)
-
- def trigger_callbacks(self):
- with self.lock:
- callbacks = self.update_callbacks[:]
- [update() for update in callbacks]
-
+ def init_up_to_date(self):
+ self.up_to_date_event.clear()
+ self.up_to_date = False
def import_key(self, keypair, password):
address, key = keypair.split(':')
return s
def get_status(self, address):
- h = self.history.get(address)
+ with self.lock:
+ h = self.history.get(address)
if not h:
status = None
else:
status = status + ':%d'% len(h)
return status
- def receive_status_callback(self, addr, status):
- with self.lock:
- if self.get_status(addr) != status:
- #print "updating status for", addr, status
- self.interface.get_history(addr)
def receive_history_callback(self, addr, data):
#print "updating history for", addr
self.save()
def get_tx_history(self):
- lines = self.tx_history.values()
+ with self.lock:
+ lines = self.tx_history.values()
lines = sorted(lines, key=operator.itemgetter("timestamp"))
return lines
+ def get_tx_hashes(self):
+ with self.lock:
+ hashes = self.tx_history.keys()
+ return hashes
+
+ def get_transactions_at_height(self, height):
+ with self.lock:
+ values = self.tx_history.values()[:]
+
+ out = []
+ for tx in values:
+ if tx['height'] == height:
+ out.append(tx['tx_hash'])
+ return out
+
def update_tx_history(self):
self.tx_history= {}
for addr in self.all_addresses():
def update(self):
- self.interface.poke()
+ self.interface.poke('synchronizer')
self.up_to_date_event.wait(10000000000)
- def start_session(self, interface):
- self.interface = interface
- self.interface.send([('server.banner',[]), ('blockchain.numblocks.subscribe',[]), ('server.peers.subscribe',[])])
- self.interface.subscribe(self.all_addresses())
-
-
def freeze(self,addr):
if addr in self.all_addresses() and addr not in self.frozen_addresses:
self.unprioritize(addr)
for k, v in s.items():
self.config.set_key(k,v)
self.config.save()
+
+
+
+
+
+
+class WalletSynchronizer(threading.Thread):
+
+
+ def __init__(self, wallet, config):
+ threading.Thread.__init__(self)
+ self.daemon = True
+ self.wallet = wallet
+ self.interface = self.wallet.interface
+ self.interface.register_channel('synchronizer')
+ self.wallet.interface.register_callback('connected', self.wallet.init_up_to_date)
+ self.wallet.interface.register_callback('connected', lambda: self.interface.send([('server.banner',[])],'synchronizer') )
+
+ def synchronize_wallet(self):
+ new_addresses = self.wallet.synchronize()
+ if new_addresses:
+ self.subscribe_to_addresses(new_addresses)
+
+ if self.interface.is_up_to_date('synchronizer'):
+ if not self.wallet.up_to_date:
+ self.wallet.up_to_date = True
+ self.wallet.was_updated = True
+ self.wallet.up_to_date_event.set()
+ else:
+ if self.wallet.up_to_date:
+ self.wallet.up_to_date = False
+ self.wallet.was_updated = True
+
+
+
+ def subscribe_to_addresses(self, addresses):
+ messages = []
+ for addr in addresses:
+ messages.append(('blockchain.address.subscribe', [addr]))
+ self.interface.send( messages, 'synchronizer')
+
+
+ def run(self):
+
+ # wait until we are connected, in case the user is not connected
+ while not self.interface.is_connected:
+ time.sleep(1)
+
+ # request banner, because 'connected' event happens before this thread is started
+ self.interface.send([('server.banner',[])],'synchronizer')
+
+ # subscriptions
+ self.interface.send([('blockchain.numblocks.subscribe',[])], 'synchronizer')
+ self.interface.send([('server.peers.subscribe',[])],'synchronizer')
+ self.subscribe_to_addresses(self.wallet.all_addresses())
+
+ while True:
+ # 1. send new requests
+ self.synchronize_wallet()
+
+ if self.wallet.was_updated:
+ self.interface.trigger_callback('updated')
+ self.wallet.was_updated = False
+
+ # 2. get a response
+ r = self.interface.get_response('synchronizer')
+ if not r: continue
+
+ # 3. handle response
+ method = r['method']
+ params = r['params']
+ result = r['result']
+
+ if method == 'blockchain.address.subscribe':
+ addr = params[0]
+ if self.wallet.get_status(addr) != result:
+ self.interface.send([('blockchain.address.get_history', [address] )])
+
+ elif method == 'blockchain.address.get_history':
+ addr = params[0]
+ self.wallet.receive_history_callback(addr, result)
+ self.wallet.was_updated = True
+
+ elif method == 'blockchain.transaction.broadcast':
+ self.wallet.tx_result = result
+ self.wallet.tx_event.set()
+
+ elif method == 'blockchain.numblocks.subscribe':
+ self.wallet.blocks = result
+ self.wallet.was_updated = True
+
+ elif method == 'server.version':
+ pass
+
+ elif method == 'server.peers.subscribe':
+ servers = []
+ for item in result:
+ s = []
+ host = item[1]
+ ports = []
+ version = None
+ if len(item) > 2:
+ for v in item[2]:
+ if re.match("[stgh]\d+", v):
+ ports.append((v[0], v[1:]))
+ if re.match("v(.?)+", v):
+ version = v[1:]
+ if ports and version:
+ servers.append((host, ports))
+ self.interface.servers = servers
+ self.interface.trigger_callback('peers')
+
+ elif method == 'server.banner':
+ self.wallet.banner = result
+ self.wallet.was_updated = True
+
+ else:
+ print_error("Error: Unknown message:" + method + ", " + repr(params) + ", " + repr(result) )
+
+ if self.wallet.was_updated:
+ self.interface.trigger_callback('updated')
+ self.wallet.was_updated = False
+
+
git://git.novaco.in / electrum-nvc.git / blobdiff