3 changed files with 233 additions and 199 deletions
@ -0,0 +1,195 @@
|
||||
#!/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 <http://www.gnu.org/licenses/>. |
||||
|
||||
|
||||
import threading |
||||
import Queue |
||||
import bitcoin |
||||
from util import print_error |
||||
from transaction import Transaction |
||||
|
||||
|
||||
class WalletSynchronizer(threading.Thread): |
||||
|
||||
def __init__(self, wallet, network): |
||||
threading.Thread.__init__(self) |
||||
self.daemon = True |
||||
self.wallet = wallet |
||||
self.network = network |
||||
self.was_updated = True |
||||
self.running = False |
||||
self.lock = threading.Lock() |
||||
self.queue = Queue.Queue() |
||||
|
||||
def stop(self): |
||||
with self.lock: self.running = False |
||||
|
||||
def is_running(self): |
||||
with self.lock: return self.running |
||||
|
||||
|
||||
def subscribe_to_addresses(self, addresses): |
||||
messages = [] |
||||
for addr in addresses: |
||||
messages.append(('blockchain.address.subscribe', [addr])) |
||||
self.network.subscribe( messages, lambda i,r: self.queue.put(r)) |
||||
|
||||
|
||||
def run(self): |
||||
with self.lock: |
||||
self.running = True |
||||
|
||||
while self.is_running(): |
||||
|
||||
if not self.network.is_connected(): |
||||
self.network.wait_until_connected() |
||||
|
||||
self.run_interface() |
||||
|
||||
|
||||
def run_interface(self): |
||||
|
||||
print_error("synchronizer: connected to", self.network.main_server()) |
||||
|
||||
requested_tx = [] |
||||
missing_tx = [] |
||||
requested_histories = {} |
||||
|
||||
# request any missing transactions |
||||
for history in self.wallet.history.values(): |
||||
if history == ['*']: continue |
||||
for tx_hash, tx_height in history: |
||||
if self.wallet.transactions.get(tx_hash) is None and (tx_hash, tx_height) not in missing_tx: |
||||
missing_tx.append( (tx_hash, tx_height) ) |
||||
|
||||
if missing_tx: |
||||
print_error("missing tx", missing_tx) |
||||
|
||||
# subscriptions |
||||
self.subscribe_to_addresses(self.wallet.addresses(True)) |
||||
|
||||
while self.is_running(): |
||||
# 1. create new addresses |
||||
new_addresses = self.wallet.synchronize() |
||||
|
||||
# request missing addresses |
||||
if new_addresses: |
||||
self.subscribe_to_addresses(new_addresses) |
||||
|
||||
# request missing transactions |
||||
for tx_hash, tx_height in missing_tx: |
||||
if (tx_hash, tx_height) not in requested_tx: |
||||
self.network.send([ ('blockchain.transaction.get',[tx_hash, tx_height]) ], lambda i,r: self.queue.put(r)) |
||||
requested_tx.append( (tx_hash, tx_height) ) |
||||
missing_tx = [] |
||||
|
||||
# detect if situation has changed |
||||
if self.network.is_up_to_date() and self.queue.empty(): |
||||
if not self.wallet.is_up_to_date(): |
||||
self.wallet.set_up_to_date(True) |
||||
self.was_updated = True |
||||
else: |
||||
if self.wallet.is_up_to_date(): |
||||
self.wallet.set_up_to_date(False) |
||||
self.was_updated = True |
||||
|
||||
if self.was_updated: |
||||
self.network.trigger_callback('updated') |
||||
self.was_updated = False |
||||
|
||||
# 2. get a response |
||||
try: |
||||
r = self.queue.get(block=True, timeout=1) |
||||
except Queue.Empty: |
||||
continue |
||||
|
||||
# see if it changed |
||||
#if interface != self.network.interface: |
||||
# break |
||||
|
||||
if not r: |
||||
continue |
||||
|
||||
# 3. handle response |
||||
method = r['method'] |
||||
params = r['params'] |
||||
result = r.get('result') |
||||
error = r.get('error') |
||||
if error: |
||||
print "error", r |
||||
continue |
||||
|
||||
if method == 'blockchain.address.subscribe': |
||||
addr = params[0] |
||||
if self.wallet.get_status(self.wallet.get_history(addr)) != result: |
||||
if requested_histories.get(addr) is None: |
||||
self.network.send([('blockchain.address.get_history', [addr])], lambda i,r:self.queue.put(r)) |
||||
requested_histories[addr] = result |
||||
|
||||
elif method == 'blockchain.address.get_history': |
||||
addr = params[0] |
||||
print_error("receiving history", addr, result) |
||||
if result == ['*']: |
||||
assert requested_histories.pop(addr) == '*' |
||||
self.wallet.receive_history_callback(addr, result) |
||||
else: |
||||
hist = [] |
||||
# check that txids are unique |
||||
txids = [] |
||||
for item in result: |
||||
tx_hash = item['tx_hash'] |
||||
if tx_hash not in txids: |
||||
txids.append(tx_hash) |
||||
hist.append( (tx_hash, item['height']) ) |
||||
|
||||
if len(hist) != len(result): |
||||
raise Exception("error: server sent history with non-unique txid", result) |
||||
|
||||
# check that the status corresponds to what was announced |
||||
rs = requested_histories.pop(addr) |
||||
if self.wallet.get_status(hist) != rs: |
||||
raise Exception("error: status mismatch: %s"%addr) |
||||
|
||||
# store received history |
||||
self.wallet.receive_history_callback(addr, hist) |
||||
|
||||
# request transactions that we don't have |
||||
for tx_hash, tx_height in hist: |
||||
if self.wallet.transactions.get(tx_hash) is None: |
||||
if (tx_hash, tx_height) not in requested_tx and (tx_hash, tx_height) not in missing_tx: |
||||
missing_tx.append( (tx_hash, tx_height) ) |
||||
|
||||
elif method == 'blockchain.transaction.get': |
||||
tx_hash = params[0] |
||||
tx_height = params[1] |
||||
assert tx_hash == bitcoin.hash_encode(bitcoin.Hash(result.decode('hex'))) |
||||
tx = Transaction(result) |
||||
self.wallet.receive_tx_callback(tx_hash, tx, tx_height) |
||||
self.was_updated = True |
||||
requested_tx.remove( (tx_hash, tx_height) ) |
||||
print_error("received tx:", tx_hash, len(tx.raw)) |
||||
|
||||
else: |
||||
print_error("Error: Unknown message:" + method + ", " + repr(params) + ", " + repr(result) ) |
||||
|
||||
if self.was_updated and not requested_tx: |
||||
self.network.trigger_callback('updated') |
||||
# Updated gets called too many times from other places as well; if we use that signal we get the notification three times |
||||
self.network.trigger_callback("new_transaction") |
||||
self.was_updated = False |
||||
|
||||
Loading…
Reference in new issue