lib/synchronizer.py

   1 #!/usr/bin/env python
   2 #
   3 # Electrum - lightweight Bitcoin client
   4 # Copyright (C) 2014 Thomas Voegtlin
   5 #
   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.
  10 #
  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.
  15 #
  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/>.
  18
  19
  20 import threading
  21 import Queue
  22 import bitcoin
  23 from util import print_error
  24 from transaction import Transaction
  25
  26
  27 class WalletSynchronizer(threading.Thread):
  28
  29     def __init__(self, wallet, network):
  30         threading.Thread.__init__(self)
  31         self.daemon = True
  32         self.wallet = wallet
  33         self.network = network
  34         self.was_updated = True
  35         self.running = False
  36         self.lock = threading.Lock()
  37         self.queue = Queue.Queue()
  38         self.address_queue = Queue.Queue()
  39
  40     def stop(self):
  41         with self.lock: self.running = False
  42
  43     def is_running(self):
  44         with self.lock: return self.running
  45
  46     def add(self, address):
  47         self.address_queue.put(address)
  48
  49     def subscribe_to_addresses(self, addresses):
  50         messages = []
  51         for addr in addresses:
  52             messages.append(('blockchain.address.subscribe', [addr]))
  53         self.network.subscribe( messages, lambda i,r: self.queue.put(r))
  54
  55     def run(self):
  56         with self.lock:
  57             self.running = True
  58         while self.is_running():
  59             if not self.network.is_connected():
  60                 self.network.wait_until_connected()
  61             self.run_interface()
  62
  63     def run_interface(self):
  64         print_error("synchronizer: connected to", self.network.main_server())
  65
  66         requested_tx = []
  67         missing_tx = []
  68         requested_histories = {}
  69
  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) )
  76
  77         if missing_tx:
  78             print_error("missing tx", missing_tx)
  79
  80         # subscriptions
  81         self.subscribe_to_addresses(self.wallet.addresses(True))
  82
  83         while self.is_running():
  84
  85             # 1. create new addresses
  86             self.wallet.synchronize()
  87
  88             # request missing addresses
  89             new_addresses = []
  90             while True:
  91                 try:
  92                     addr = self.address_queue.get(block=False)
  93                 except Queue.Empty:
  94                     break
  95                 new_addresses.append(addr)
  96             if new_addresses:
  97                 self.subscribe_to_addresses(new_addresses)
  98
  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) )
 104             missing_tx = []
 105
 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
 111             else:
 112                 if self.wallet.is_up_to_date():
 113                     self.wallet.set_up_to_date(False)
 114                     self.was_updated = True
 115
 116             if self.was_updated:
 117                 self.network.trigger_callback('updated')
 118                 self.was_updated = False
 119
 120             # 2. get a response
 121             try:
 122                 r = self.queue.get(block=True, timeout=1)
 123             except Queue.Empty:
 124                 continue
 125
 126             # see if it changed
 127             #if interface != self.network.interface:
 128             #    break
 129
 130             if not r:
 131                 continue
 132
 133             # 3. handle response
 134             method = r['method']
 135             params = r['params']
 136             result = r.get('result')
 137             error = r.get('error')
 138             if error:
 139                 print "error", r
 140                 continue
 141
 142             if method == 'blockchain.address.subscribe':
 143                 addr = params[0]
 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
 148
 149             elif method == 'blockchain.address.get_history':
 150                 addr = params[0]
 151                 print_error("receiving history", addr, result)
 152                 if result == ['*']:
 153                     assert requested_histories.pop(addr) == '*'
 154                     self.wallet.receive_history_callback(addr, result)
 155                 else:
 156                     hist = []
 157                     # check that txids are unique
 158                     txids = []
 159                     for item in result:
 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']) )
 164
 165                     if len(hist) != len(result):
 166                         raise Exception("error: server sent history with non-unique txid", result)
 167
 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)
 172
 173                     # store received history
 174                     self.wallet.receive_history_callback(addr, hist)
 175
 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) )
 181
 182             elif method == 'blockchain.transaction.get':
 183                 tx_hash = params[0]
 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))
 191
 192             else:
 193                 print_error("Error: Unknown message:" + method + ", " + repr(params) + ", " + repr(result) )
 194
 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
 200