lib/interface.py

   1 #!/usr/bin/env python
   2 #
   3 # Electrum - lightweight Bitcoin client
   4 # Copyright (C) 2011 thomasv@gitorious
   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 random, socket, ast, re
  21 import threading, traceback, sys, time, json, Queue
  22
  23 from version import ELECTRUM_VERSION
  24
  25 DEFAULT_TIMEOUT = 5
  26 DEFAULT_SERVERS = [ 'ecdsa.org:50001:t', 'electrum.novit.ro:50001:t', 'electrum.bitcoins.sk:50001:t']  # list of default servers
  27
  28
  29 def old_to_new(s):
  30     s = s.replace("'blk_hash'", "'block_hash'")
  31     s = s.replace("'pos'", "'index'")
  32     s = s.replace("'nTime'", "'timestamp'")
  33     s = s.replace("'is_in'", "'is_input'")
  34     s = s.replace("'raw_scriptPubKey'","'raw_output_script'")
  35     return s
  36
  37
  38 class Interface(threading.Thread):
  39     def __init__(self, host, port):
  40         threading.Thread.__init__(self)
  41         self.daemon = True
  42         self.host = host
  43         self.port = port
  44
  45         self.servers = [] # actual list from IRC
  46         self.rtime = 0
  47
  48         self.is_connected = True
  49         self.poll_interval = 1
  50
  51         #json
  52         self.message_id = 0
  53         self.responses = Queue.Queue()
  54         self.unanswered_requests = {}
  55
  56     def init_socket(self):
  57         pass
  58
  59     def poke(self):
  60         # push a fake response so that the getting thread exits its loop
  61         self.responses.put(None)
  62
  63     def queue_json_response(self, c):
  64
  65         #print "<--",c
  66         msg_id = c.get('id')
  67         error = c.get('error')
  68
  69         if error:
  70             print "received error:", c
  71             return
  72
  73         if msg_id is not None:
  74             method, params = self.unanswered_requests.pop(msg_id)
  75             result = c.get('result')
  76         else:
  77             # notification
  78             method = c.get('method')
  79             params = c.get('params')
  80
  81             if method == 'blockchain.numblocks.subscribe':
  82                 result = params[0]
  83                 params = []
  84
  85             elif method == 'blockchain.address.subscribe':
  86                 addr = params[0]
  87                 result = params[1]
  88                 params = [addr]
  89
  90         self.responses.put({'method':method, 'params':params, 'result':result})
  91
  92
  93
  94     def subscribe(self, addresses):
  95         messages = []
  96         for addr in addresses:
  97             messages.append(('blockchain.address.subscribe', [addr]))
  98         self.send(messages)
  99
 100
 101
 102
 103 class PollingInterface(Interface):
 104     """ non-persistent connection. synchronous calls"""
 105
 106     def __init__(self, host, port):
 107         Interface.__init__(self, host, port)
 108         self.session_id = None
 109
 110     def get_history(self, address):
 111         self.send([('blockchain.address.get_history', [address] )])
 112
 113     def poll(self):
 114         pass
 115         #if is_new or wallet.remote_url:
 116         #    self.was_updated = True
 117         #    is_new = wallet.synchronize()
 118         #    wallet.update_tx_history()
 119         #    wallet.save()
 120         #    return is_new
 121         #else:
 122         #    return False
 123
 124     def run(self):
 125         self.is_connected = True
 126         while self.is_connected:
 127             try:
 128                 if self.session_id:
 129                     self.poll()
 130                 time.sleep(self.poll_interval)
 131             except socket.gaierror:
 132                 break
 133             except socket.error:
 134                 break
 135             except:
 136                 traceback.print_exc(file=sys.stdout)
 137                 break
 138
 139         self.is_connected = False
 140         self.poke()
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150 class HttpStratumInterface(PollingInterface):
 151
 152     def poll(self):
 153         self.send([])
 154
 155     def send(self, messages):
 156         import urllib2, json, time, cookielib
 157
 158         cj = cookielib.CookieJar()
 159         opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
 160         urllib2.install_opener(opener)
 161
 162         t1 = time.time()
 163
 164         data = []
 165         for m in messages:
 166             method, params = m
 167             if type(params) != type([]): params = [params]
 168             data.append( { 'method':method, 'id':self.message_id, 'params':params } )
 169             self.unanswered_requests[self.message_id] = method, params
 170             self.message_id += 1
 171
 172         if data:
 173             data_json = json.dumps(data)
 174         else:
 175             # poll with GET
 176             data_json = None
 177
 178         host = 'http://%s:%d'%( self.host, self.port )
 179         headers = {'content-type': 'application/json'}
 180         if self.session_id:
 181             headers['cookie'] = 'SESSION=%s'%self.session_id
 182
 183         req = urllib2.Request(host, data_json, headers)
 184         response_stream = urllib2.urlopen(req)
 185
 186         for index, cookie in enumerate(cj):
 187             if cookie.name=='SESSION':
 188                 self.session_id = cookie.value
 189
 190         response = response_stream.read()
 191         if response:
 192             response = json.loads( response )
 193             if type(response) is not type([]):
 194                 self.queue_json_response(response)
 195             else:
 196                 for item in response:
 197                     self.queue_json_response(item)
 198
 199         if response:
 200             self.poll_interval = 1
 201         else:
 202             if self.poll_interval < 15:
 203                 self.poll_interval += 1
 204         #print self.poll_interval, response
 205
 206         self.rtime = time.time() - t1
 207         self.is_connected = True
 208
 209
 210
 211
 212 class TcpStratumInterface(Interface):
 213     """json-rpc over persistent TCP connection, asynchronous"""
 214
 215     def __init__(self, host, port):
 216         Interface.__init__(self, host, port)
 217
 218     def init_socket(self):
 219         self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
 220         self.s.settimeout(5*60)
 221         self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 222         try:
 223             self.s.connect(( self.host, self.port))
 224             self.is_connected = True
 225             self.send([('server.version', [ELECTRUM_VERSION])])
 226             print "Connected to %s:%d"%(self.host,self.port)
 227         except:
 228             self.is_connected = False
 229             print "Not connected"
 230
 231     def run(self):
 232         try:
 233             out = ''
 234             while self.is_connected:
 235                 try: msg = self.s.recv(1024)
 236                 except socket.timeout:
 237                     # ping the server with server.version, as a real ping does not exist yet
 238                     self.send([('server.version', [ELECTRUM_VERSION])])
 239                     continue
 240                 out += msg
 241                 if msg == '':
 242                     self.is_connected = False
 243                     print "disconnected."
 244
 245                 while True:
 246                     s = out.find('\n')
 247                     if s==-1: break
 248                     c = out[0:s]
 249                     out = out[s+1:]
 250                     c = json.loads(c)
 251                     self.queue_json_response(c)
 252
 253         except:
 254             traceback.print_exc(file=sys.stdout)
 255
 256         self.is_connected = False
 257         print "poking"
 258         self.poke()
 259
 260     def send(self, messages):
 261         out = ''
 262         for m in messages:
 263             method, params = m
 264             request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
 265             self.unanswered_requests[self.message_id] = method, params
 266             #print "-->",request
 267             self.message_id += 1
 268             out += request + '\n'
 269         self.s.send( out )
 270
 271     def get_history(self, addr):
 272         self.send([('blockchain.address.get_history', [addr])])
 273
 274
 275
 276
 277
 278 class WalletSynchronizer(threading.Thread):
 279
 280     def __init__(self, wallet, loop=False):
 281         threading.Thread.__init__(self)
 282         self.daemon = True
 283         self.wallet = wallet
 284         self.loop = loop
 285         self.init_interface()
 286
 287     def init_interface(self):
 288         try:
 289             host, port, protocol = self.wallet.server.split(':')
 290             port = int(port)
 291         except:
 292             self.wallet.pick_random_server()
 293             host, port, protocol = self.wallet.server.split(':')
 294             port = int(port)
 295
 296         #print protocol, host, port
 297         if protocol == 't':
 298             InterfaceClass = TcpStratumInterface
 299         elif protocol == 'h':
 300             InterfaceClass = HttpStratumInterface
 301         else:
 302             print "unknown protocol"
 303             InterfaceClass = TcpStratumInterface
 304
 305         self.interface = InterfaceClass(host, port)
 306         self.wallet.interface = self.interface
 307
 308
 309     def handle_response(self, r):
 310         if r is None:
 311             return
 312
 313         method = r['method']
 314         params = r['params']
 315         result = r['result']
 316
 317         if method == 'server.banner':
 318             self.wallet.banner = result
 319             self.wallet.was_updated = True
 320
 321         elif method == 'server.peers.subscribe':
 322             servers = []
 323             for item in result:
 324                 s = []
 325                 host = item[1]
 326                 ports = []
 327                 if len(item)>2:
 328                     for v in item[2]:
 329                         if re.match("[th]\d+",v):
 330                             ports.append((v[0],v[1:]))
 331                 if ports:
 332                     servers.append( (host, ports) )
 333             self.interface.servers = servers
 334
 335         elif method == 'blockchain.address.subscribe':
 336             addr = params[0]
 337             self.wallet.receive_status_callback(addr, result)
 338
 339         elif method == 'blockchain.address.get_history':
 340             addr = params[0]
 341             self.wallet.receive_history_callback(addr, result)
 342             self.wallet.was_updated = True
 343
 344         elif method == 'blockchain.transaction.broadcast':
 345             self.wallet.tx_result = result
 346             self.wallet.tx_event.set()
 347
 348         elif method == 'blockchain.numblocks.subscribe':
 349             self.wallet.blocks = result
 350
 351         elif method == 'server.version':
 352             pass
 353
 354         else:
 355             print "unknown message:", method, params, result
 356
 357
 358     def start_interface(self):
 359         self.interface.init_socket()
 360         self.interface.start()
 361         if self.interface.is_connected:
 362             self.wallet.start_session(self.interface)
 363
 364
 365
 366     def run(self):
 367         import socket, time
 368         self.start_interface()
 369         while True:
 370             while self.interface.is_connected:
 371                 new_addresses = self.wallet.synchronize()
 372                 if new_addresses:
 373                     self.interface.subscribe(new_addresses)
 374
 375                 if self.wallet.is_up_to_date():
 376                     if not self.wallet.up_to_date:
 377                         self.wallet.up_to_date = True
 378                         self.wallet.was_updated = True
 379                         self.wallet.up_to_date_event.set()
 380                 else:
 381                     if self.wallet.up_to_date:
 382                         self.wallet.up_to_date = False
 383                         self.wallet.was_updated = True
 384
 385                 if self.wallet.was_updated:
 386                     self.wallet.gui_callback()
 387                     self.wallet.was_updated = False
 388
 389                 response = self.interface.responses.get()
 390                 self.handle_response(response)
 391
 392             self.wallet.gui_callback()
 393             if self.loop:
 394                 time.sleep(5)
 395                 self.init_interface()
 396                 self.start_interface()
 397                 continue
 398             else:
 399                 break
 400
 401
 402