3 # Electrum - lightweight Bitcoin client
4 # Copyright (C) 2014 Thomas Voegtlin
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
23 from util import print_error
24 from transaction import Transaction
27 class WalletSynchronizer(threading.Thread):
29 def __init__(self, wallet, network):
30 threading.Thread.__init__(self)
33 self.network = network
34 self.was_updated = True
36 self.lock = threading.Lock()
37 self.queue = Queue.Queue()
38 self.address_queue = Queue.Queue()
41 with self.lock: self.running = False
44 with self.lock: return self.running
46 def add(self, address):
47 self.address_queue.put(address)
49 def subscribe_to_addresses(self, addresses):
51 for addr in addresses:
52 messages.append(('blockchain.address.subscribe', [addr]))
53 self.network.subscribe( messages, lambda i,r: self.queue.put(r))
58 while self.is_running():
59 if not self.network.is_connected():
60 self.network.wait_until_connected()
63 def run_interface(self):
64 print_error("synchronizer: connected to", self.network.main_server())
68 requested_histories = {}
70 # request any missing transactions
71 for history in self.wallet.history.values():
72 if history == ['*']: continue
73 for tx_hash, tx_height in history:
74 if self.wallet.transactions.get(tx_hash) is None and (tx_hash, tx_height) not in missing_tx:
75 missing_tx.append( (tx_hash, tx_height) )
78 print_error("missing tx", missing_tx)
81 self.subscribe_to_addresses(self.wallet.addresses(True))
83 while self.is_running():
85 # 1. create new addresses
86 self.wallet.synchronize()
88 # request missing addresses
92 addr = self.address_queue.get(block=False)
95 new_addresses.append(addr)
97 self.subscribe_to_addresses(new_addresses)
99 # request missing transactions
100 for tx_hash, tx_height in missing_tx:
101 if (tx_hash, tx_height) not in requested_tx:
102 self.network.send([ ('blockchain.transaction.get',[tx_hash, tx_height]) ], lambda i,r: self.queue.put(r))
103 requested_tx.append( (tx_hash, tx_height) )
106 # detect if situation has changed
107 if self.network.is_up_to_date() and self.queue.empty():
108 if not self.wallet.is_up_to_date():
109 self.wallet.set_up_to_date(True)
110 self.was_updated = True
112 if self.wallet.is_up_to_date():
113 self.wallet.set_up_to_date(False)
114 self.was_updated = True
117 self.network.trigger_callback('updated')
118 self.was_updated = False
122 r = self.queue.get(block=True, timeout=1)
127 #if interface != self.network.interface:
136 result = r.get('result')
137 error = r.get('error')
142 if method == 'blockchain.address.subscribe':
144 if self.wallet.get_status(self.wallet.get_history(addr)) != result:
145 if requested_histories.get(addr) is None:
146 self.network.send([('blockchain.address.get_history', [addr])], lambda i,r:self.queue.put(r))
147 requested_histories[addr] = result
149 elif method == 'blockchain.address.get_history':
151 print_error("receiving history", addr, result)
153 assert requested_histories.pop(addr) == '*'
154 self.wallet.receive_history_callback(addr, result)
157 # check that txids are unique
160 tx_hash = item['tx_hash']
161 if tx_hash not in txids:
162 txids.append(tx_hash)
163 hist.append( (tx_hash, item['height']) )
165 if len(hist) != len(result):
166 raise Exception("error: server sent history with non-unique txid", result)
168 # check that the status corresponds to what was announced
169 rs = requested_histories.pop(addr)
170 if self.wallet.get_status(hist) != rs:
171 raise Exception("error: status mismatch: %s"%addr)
173 # store received history
174 self.wallet.receive_history_callback(addr, hist)
176 # request transactions that we don't have
177 for tx_hash, tx_height in hist:
178 if self.wallet.transactions.get(tx_hash) is None:
179 if (tx_hash, tx_height) not in requested_tx and (tx_hash, tx_height) not in missing_tx:
180 missing_tx.append( (tx_hash, tx_height) )
182 elif method == 'blockchain.transaction.get':
184 tx_height = params[1]
185 assert tx_hash == bitcoin.hash_encode(bitcoin.Hash(result.decode('hex')))
186 tx = Transaction(result)
187 self.wallet.receive_tx_callback(tx_hash, tx, tx_height)
188 self.was_updated = True
189 requested_tx.remove( (tx_hash, tx_height) )
190 print_error("received tx:", tx_hash, len(tx.raw))
193 print_error("Error: Unknown message:" + method + ", " + repr(params) + ", " + repr(result) )
195 if self.was_updated and not requested_tx:
196 self.network.trigger_callback('updated')
197 # Updated gets called too many times from other places as well; if we use that signal we get the notification three times
198 self.network.trigger_callback("new_transaction")
199 self.was_updated = False