3 # Electrum - lightweight Bitcoin client
4 # Copyright (C) 2011 thomasv@gitorious
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/>.
20 import random, socket, ast, re
21 import threading, traceback, sys, time, json, Queue
23 from version import ELECTRUM_VERSION
26 DEFAULT_SERVERS = [ 'ecdsa.org:50001:t', 'electrum.novit.ro:50001:t', 'electrum.bitcoins.sk:50001:t'] # list of default servers
29 def replace_keys(obj, old_key, new_key):
30 if isinstance(obj, dict):
32 obj[new_key] = obj[old_key]
34 for elem in obj.itervalues():
35 replace_keys(elem, old_key, new_key)
36 elif isinstance(obj, list):
38 replace_keys(elem, old_key, new_key)
41 replace_keys(d, 'blk_hash', 'block_hash')
42 replace_keys(d, 'pos', 'index')
43 replace_keys(d, 'nTime', 'timestamp')
44 replace_keys(d, 'is_in', 'is_input')
45 replace_keys(d, 'raw_scriptPubKey', 'raw_output_script')
48 class Interface(threading.Thread):
49 def __init__(self, host, port):
50 threading.Thread.__init__(self)
55 self.servers = [] # actual list from IRC
57 self.bytes_received = 0
59 self.is_connected = True
60 self.poll_interval = 1
64 self.responses = Queue.Queue()
65 self.unanswered_requests = {}
67 def init_socket(self):
71 # push a fake response so that the getting thread exits its loop
72 self.responses.put(None)
74 def queue_json_response(self, c):
78 error = c.get('error')
81 print "received error:", c
84 if msg_id is not None:
85 method, params = self.unanswered_requests.pop(msg_id)
86 result = c.get('result')
89 method = c.get('method')
90 params = c.get('params')
92 if method == 'blockchain.numblocks.subscribe':
96 elif method == 'blockchain.address.subscribe':
101 self.responses.put({'method':method, 'params':params, 'result':result})
105 def subscribe(self, addresses):
107 for addr in addresses:
108 messages.append(('blockchain.address.subscribe', [addr]))
114 class PollingInterface(Interface):
115 """ non-persistent connection. synchronous calls"""
117 def __init__(self, host, port):
118 Interface.__init__(self, host, port)
119 self.session_id = None
121 def get_history(self, address):
122 self.send([('blockchain.address.get_history', [address] )])
126 #if is_new or wallet.remote_url:
127 # self.was_updated = True
128 # is_new = wallet.synchronize()
129 # wallet.update_tx_history()
136 self.is_connected = True
137 while self.is_connected:
141 time.sleep(self.poll_interval)
142 except socket.gaierror:
147 traceback.print_exc(file=sys.stdout)
150 self.is_connected = False
161 class HttpStratumInterface(PollingInterface):
166 def send(self, messages):
167 import urllib2, json, time, cookielib
169 cj = cookielib.CookieJar()
170 opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
171 urllib2.install_opener(opener)
178 if type(params) != type([]): params = [params]
179 data.append( { 'method':method, 'id':self.message_id, 'params':params } )
180 self.unanswered_requests[self.message_id] = method, params
184 data_json = json.dumps(data)
189 host = 'http://%s:%d'%( self.host, self.port )
190 headers = {'content-type': 'application/json'}
192 headers['cookie'] = 'SESSION=%s'%self.session_id
194 req = urllib2.Request(host, data_json, headers)
195 response_stream = urllib2.urlopen(req)
197 for index, cookie in enumerate(cj):
198 if cookie.name=='SESSION':
199 self.session_id = cookie.value
201 response = response_stream.read()
202 self.bytes_received += len(response)
204 response = json.loads( response )
205 if type(response) is not type([]):
206 self.queue_json_response(response)
208 for item in response:
209 self.queue_json_response(item)
212 self.poll_interval = 1
214 if self.poll_interval < 15:
215 self.poll_interval += 1
216 #print self.poll_interval, response
218 self.rtime = time.time() - t1
219 self.is_connected = True
224 class TcpStratumInterface(Interface):
225 """json-rpc over persistent TCP connection, asynchronous"""
227 def __init__(self, host, port):
228 Interface.__init__(self, host, port)
230 def init_socket(self):
231 self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
232 self.s.settimeout(60)
233 self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
235 self.s.connect(( self.host, self.port))
236 self.is_connected = True
237 self.send([('server.version', [ELECTRUM_VERSION])])
238 print "Connected to %s:%d"%(self.host,self.port)
240 self.is_connected = False
241 print "Not connected"
246 while self.is_connected:
247 try: msg = self.s.recv(1024)
248 except socket.timeout:
249 # ping the server with server.version, as a real ping does not exist yet
250 self.send([('server.version', [ELECTRUM_VERSION])])
253 self.bytes_received += len(msg)
255 self.is_connected = False
256 print "disconnected."
264 self.queue_json_response(c)
267 traceback.print_exc(file=sys.stdout)
269 self.is_connected = False
273 def send(self, messages):
277 request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
278 self.unanswered_requests[self.message_id] = method, params
281 out += request + '\n'
284 sent = self.s.send( out )
287 def get_history(self, addr):
288 self.send([('blockchain.address.get_history', [addr])])
294 class WalletSynchronizer(threading.Thread):
296 def __init__(self, wallet, loop=False):
297 threading.Thread.__init__(self)
301 self.init_interface()
303 def init_interface(self):
305 host, port, protocol = self.wallet.server.split(':')
308 self.wallet.pick_random_server()
309 host, port, protocol = self.wallet.server.split(':')
312 #print protocol, host, port
314 InterfaceClass = TcpStratumInterface
315 elif protocol == 'h':
316 InterfaceClass = HttpStratumInterface
318 print "unknown protocol"
319 InterfaceClass = TcpStratumInterface
321 self.interface = InterfaceClass(host, port)
322 self.wallet.interface = self.interface
325 def handle_response(self, r):
333 if method == 'server.banner':
334 self.wallet.banner = result
335 self.wallet.was_updated = True
337 elif method == 'server.peers.subscribe':
345 if re.match("[th]\d+",v):
346 ports.append((v[0],v[1:]))
348 servers.append( (host, ports) )
349 self.interface.servers = servers
351 elif method == 'blockchain.address.subscribe':
353 self.wallet.receive_status_callback(addr, result)
355 elif method == 'blockchain.address.get_history':
357 self.wallet.receive_history_callback(addr, result)
358 self.wallet.was_updated = True
360 elif method == 'blockchain.transaction.broadcast':
361 self.wallet.tx_result = result
362 self.wallet.tx_event.set()
364 elif method == 'blockchain.numblocks.subscribe':
365 self.wallet.blocks = result
366 self.wallet.was_updated = True
368 elif method == 'server.version':
372 print "unknown message:", method, params, result
375 def start_interface(self):
376 self.interface.init_socket()
377 self.interface.start()
378 if self.interface.is_connected:
379 self.wallet.start_session(self.interface)
385 self.start_interface()
387 while self.interface.is_connected:
388 new_addresses = self.wallet.synchronize()
390 self.interface.subscribe(new_addresses)
392 if self.wallet.is_up_to_date():
393 if not self.wallet.up_to_date:
394 self.wallet.up_to_date = True
395 self.wallet.was_updated = True
396 self.wallet.up_to_date_event.set()
398 if self.wallet.up_to_date:
399 self.wallet.up_to_date = False
400 self.wallet.was_updated = True
402 if self.wallet.was_updated:
403 self.wallet.gui_callback()
404 self.wallet.was_updated = False
406 response = self.interface.responses.get()
407 self.handle_response(response)
409 self.wallet.gui_callback()
412 self.init_interface()
413 self.start_interface()