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
  39     def stop(self):
  40         with self.lock: self.running = False
  41
  42     def is_running(self):
  43         with self.lock: return self.running
  44
  45
  46     def subscribe_to_addresses(self, addresses):
  47         messages = []
  48         for addr in addresses:
  49             messages.append(('blockchain.address.subscribe', [addr]))
  50         self.network.subscribe( messages, lambda i,r: self.queue.put(r))
  51
  52
  53     def run(self):
  54         with self.lock:
  55             self.running = True
  56
  57         while self.is_running():
  58
  59             if not self.network.is_connected():
  60                 self.network.wait_until_connected()
  61
  62             self.run_interface()
  63
  64
  65     def run_interface(self):
  66
  67         print_error("synchronizer: connected to", self.network.main_server())
  68
  69         requested_tx = []
  70         missing_tx = []
  71         requested_histories = {}
  72
  73         # request any missing transactions
  74         for history in self.wallet.history.values():
  75             if history == ['*']: continue
  76             for tx_hash, tx_height in history:
  77                 if self.wallet.transactions.get(tx_hash) is None and (tx_hash, tx_height) not in missing_tx:
  78                     missing_tx.append( (tx_hash, tx_height) )
  79
  80         if missing_tx:
  81             print_error("missing tx", missing_tx)
  82
  83         # subscriptions
  84         self.subscribe_to_addresses(self.wallet.addresses(True))
  85
  86         while self.is_running():
  87             # 1. create new addresses
  88             new_addresses = self.wallet.synchronize()
  89
  90             # request missing addresses
  91             if new_addresses:
  92                 self.subscribe_to_addresses(new_addresses)
  93
  94             # request missing transactions
  95             for tx_hash, tx_height in missing_tx:
  96                 if (tx_hash, tx_height) not in requested_tx:
  97                     self.network.send([ ('blockchain.transaction.get',[tx_hash, tx_height]) ], lambda i,r: self.queue.put(r))
  98                     requested_tx.append( (tx_hash, tx_height) )
  99             missing_tx = []
 100
 101             # detect if situation has changed
 102             if self.network.is_up_to_date() and self.queue.empty():
 103                 if not self.wallet.is_up_to_date():
 104                     self.wallet.set_up_to_date(True)
 105                     self.was_updated = True
 106             else:
 107                 if self.wallet.is_up_to_date():
 108                     self.wallet.set_up_to_date(False)
 109                     self.was_updated = True
 110
 111             if self.was_updated:
 112                 self.network.trigger_callback('updated')
 113                 self.was_updated = False
 114
 115             # 2. get a response
 116             try:
 117                 r = self.queue.get(block=True, timeout=1)
 118             except Queue.Empty:
 119                 continue
 120
 121             # see if it changed
 122             #if interface != self.network.interface:
 123             #    break
 124
 125             if not r:
 126                 continue
 127
 128             # 3. handle response
 129             method = r['method']
 130             params = r['params']
 131             result = r.get('result')
 132             error = r.get('error')
 133             if error:
 134                 print "error", r
 135                 continue
 136
 137             if method == 'blockchain.address.subscribe':
 138                 addr = params[0]
 139                 if self.wallet.get_status(self.wallet.get_history(addr)) != result:
 140                     if requested_histories.get(addr) is None:
 141                         self.network.send([('blockchain.address.get_history', [addr])], lambda i,r:self.queue.put(r))
 142                         requested_histories[addr] = result
 143
 144             elif method == 'blockchain.address.get_history':
 145                 addr = params[0]
 146                 print_error("receiving history", addr, result)
 147                 if result == ['*']:
 148                     assert requested_histories.pop(addr) == '*'
 149                     self.wallet.receive_history_callback(addr, result)
 150                 else:
 151                     hist = []
 152                     # check that txids are unique
 153                     txids = []
 154                     for item in result:
 155                         tx_hash = item['tx_hash']
 156                         if tx_hash not in txids:
 157                             txids.append(tx_hash)
 158                             hist.append( (tx_hash, item['height']) )
 159
 160                     if len(hist) != len(result):
 161                         raise Exception("error: server sent history with non-unique txid", result)
 162
 163                     # check that the status corresponds to what was announced
 164                     rs = requested_histories.pop(addr)
 165                     if self.wallet.get_status(hist) != rs:
 166                         raise Exception("error: status mismatch: %s"%addr)
 167
 168                     # store received history
 169                     self.wallet.receive_history_callback(addr, hist)
 170
 171                     # request transactions that we don't have
 172                     for tx_hash, tx_height in hist:
 173                         if self.wallet.transactions.get(tx_hash) is None:
 174                             if (tx_hash, tx_height) not in requested_tx and (tx_hash, tx_height) not in missing_tx:
 175                                 missing_tx.append( (tx_hash, tx_height) )
 176
 177             elif method == 'blockchain.transaction.get':
 178                 tx_hash = params[0]
 179                 tx_height = params[1]
 180                 assert tx_hash == bitcoin.hash_encode(bitcoin.Hash(result.decode('hex')))
 181                 tx = Transaction(result)
 182                 self.wallet.receive_tx_callback(tx_hash, tx, tx_height)
 183                 self.was_updated = True
 184                 requested_tx.remove( (tx_hash, tx_height) )
 185                 print_error("received tx:", tx_hash, len(tx.raw))
 186
 187             else:
 188                 print_error("Error: Unknown message:" + method + ", " + repr(params) + ", " + repr(result) )
 189
 190             if self.was_updated and not requested_tx:
 191                 self.network.trigger_callback('updated')
 192                 # Updated gets called too many times from other places as well; if we use that signal we get the notification three times
 193                 self.network.trigger_callback("new_transaction")
 194                 self.was_updated = False
 195