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
24 from util import print_error
28 DEFAULT_SERVERS = [ 'electrum.novit.ro:50001:t',
29 'electrum.bytesized-hosting.com:50001:t'] # list of default servers
31 proxy_modes = ['socks4', 'socks5', 'http']
33 def pick_random_server():
34 print "using random server"
35 return random.choice( DEFAULT_SERVERS )
42 class InterfaceAncestor(threading.Thread):
44 def __init__(self, host, port, proxy=None):
45 threading.Thread.__init__(self)
51 self.servers = [] # actual list from IRC
53 self.bytes_received = 0
55 self.is_connected = True
56 self.poll_interval = 1
60 self.responses = Queue.Queue()
61 self.unanswered_requests = {}
65 # push a fake response so that the getting thread exits its loop
66 self.responses.put(None)
68 def queue_json_response(self, c):
74 error = c.get('error')
77 print "received error:", c
80 if msg_id is not None:
81 method, params = self.unanswered_requests.pop(msg_id)
82 result = c.get('result')
85 method = c.get('method')
86 params = c.get('params')
88 if method == 'blockchain.numblocks.subscribe':
92 elif method == 'blockchain.address.subscribe':
97 self.responses.put({'method':method, 'params':params, 'result':result})
101 def subscribe(self, addresses):
103 for addr in addresses:
104 messages.append(('blockchain.address.subscribe', [addr]))
111 class HttpStratumInterface(InterfaceAncestor):
112 """ non-persistent connection. synchronous calls"""
114 def __init__(self, host, port, proxy=None):
115 InterfaceAncestor.__init__(self, host, port, proxy)
116 self.session_id = None
118 def get_history(self, address):
119 self.send([('blockchain.address.get_history', [address] )])
122 self.is_connected = True
123 while self.is_connected:
127 time.sleep(self.poll_interval)
128 except socket.gaierror:
133 traceback.print_exc(file=sys.stdout)
136 self.is_connected = False
143 def send(self, messages):
144 import urllib2, json, time, cookielib
148 socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
149 socks.wrapmodule(urllib2)
151 cj = cookielib.CookieJar()
152 opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
153 urllib2.install_opener(opener)
160 if type(params) != type([]): params = [params]
161 data.append( { 'method':method, 'id':self.message_id, 'params':params } )
162 self.unanswered_requests[self.message_id] = method, params
166 data_json = json.dumps(data)
171 host = 'http://%s:%d'%( self.host, self.port )
172 headers = {'content-type': 'application/json'}
174 headers['cookie'] = 'SESSION=%s'%self.session_id
176 req = urllib2.Request(host, data_json, headers)
177 response_stream = urllib2.urlopen(req)
179 for index, cookie in enumerate(cj):
180 if cookie.name=='SESSION':
181 self.session_id = cookie.value
183 response = response_stream.read()
184 self.bytes_received += len(response)
186 response = json.loads( response )
187 if type(response) is not type([]):
188 self.queue_json_response(response)
190 for item in response:
191 self.queue_json_response(item)
194 self.poll_interval = 1
196 if self.poll_interval < 15:
197 self.poll_interval += 1
198 #print self.poll_interval, response
200 self.rtime = time.time() - t1
201 self.is_connected = True
206 class TcpStratumInterface(InterfaceAncestor):
207 """json-rpc over persistent TCP connection, asynchronous"""
209 def __init__(self, host, port, proxy=None):
210 InterfaceAncestor.__init__(self, host, port, proxy)
213 def init_socket(self):
215 connection_msg = "%s:%d"%(self.host,self.port)
216 if self.proxy is None:
217 self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
219 connection_msg += " using proxy %s:%s:%s"%(self.proxy.get('mode'), self.proxy.get('host'), self.proxy.get('port'))
221 self.s = socks.socksocket()
222 self.s.setproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
223 self.s.settimeout(60)
224 self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
226 self.s.connect(( self.host.encode('ascii'), int(self.port)))
227 self.is_connected = True
228 self.send([('server.version', [ELECTRUM_VERSION])])
229 print "Connected to " + connection_msg
231 self.is_connected = False
232 print_error("Failed to connect" + connection_msg)
237 while self.is_connected:
238 try: msg = self.s.recv(1024)
239 except socket.timeout:
240 # ping the server with server.version, as a real ping does not exist yet
241 self.send([('server.version', [ELECTRUM_VERSION])])
244 self.bytes_received += len(msg)
246 self.is_connected = False
247 print "Disconnected."
255 self.queue_json_response(c)
258 traceback.print_exc(file=sys.stdout)
260 self.is_connected = False
264 def send(self, messages):
268 request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
269 self.unanswered_requests[self.message_id] = method, params
272 # print "-->",request
275 out += request + '\n'
278 sent = self.s.send( out )
281 def get_history(self, addr):
282 self.send([('blockchain.address.get_history', [addr])])
286 class Interface(TcpStratumInterface, HttpStratumInterface):
288 def __init__(self, config = None):
291 from simple_config import SimpleConfig
292 config = SimpleConfig()
295 s = config.get('server')
296 host, port, protocol = s.split(':')
299 s = pick_random_server()
300 host, port, protocol = s.split(':')
303 self.protocol = protocol
304 proxy = self.parse_proxy_options(config.get('proxy'))
305 self.server = host + ':%d:%s'%(port, protocol)
307 #print protocol, host, port
309 TcpStratumInterface.__init__(self, host, port, proxy)
310 elif protocol == 'h':
311 HttpStratumInterface.__init__(self, host, port, proxy)
313 print_error("Error: Unknown protocol")
314 TcpStratumInterface.__init__(self, host, port, proxy)
318 if self.protocol == 't':
319 TcpStratumInterface.run(self)
321 HttpStratumInterface.run(self)
323 def send(self, messages):
324 if self.protocol == 't':
325 TcpStratumInterface.send(self, messages)
327 HttpStratumInterface.send(self, messages)
330 def parse_proxy_options(self, s):
331 if type(s) == type({}): return s # fixme: type should be fixed
332 if type(s) != type(""): return None
333 if s.lower() == 'none': return None
334 proxy = { "mode":"socks5", "host":"localhost" }
337 if proxy_modes.count(args[n]) == 1:
338 proxy["mode"] = args[n]
341 proxy["host"] = args[n]
344 proxy["port"] = args[n]
346 proxy["port"] = "8080" if proxy["mode"] == "http" else "1080"
350 def set_server(self, server, proxy=None):
351 # raise an error if the format isnt correct
352 a,b,c = server.split(':')
354 assert c in ['t', 'h']
356 if server != self.server or proxy != self.proxy:
357 print "changing server:", server, proxy
360 self.is_connected = False # this exits the polling loop
368 class WalletSynchronizer(threading.Thread):
370 def __init__(self, wallet, config, loop=False, servers_loaded_callback=None):
371 threading.Thread.__init__(self)
376 self.init_interface()
377 self.servers_loaded_callback = servers_loaded_callback
379 def init_interface(self):
380 self.interface = Interface(self.config)
381 self.wallet.interface = self.interface
383 def handle_response(self, r):
391 if method == 'server.banner':
392 self.wallet.banner = result
393 self.wallet.was_updated = True
395 elif method == 'server.peers.subscribe':
404 if re.match("[th]\d+", v):
405 ports.append((v[0], v[1:]))
406 if re.match("v(.?)+", v):
408 if ports and version:
409 servers.append((host, ports))
410 self.interface.servers = servers
411 # servers_loaded_callback is None for commands, but should
412 # NEVER be None when using the GUI.
413 if self.servers_loaded_callback is not None:
414 self.servers_loaded_callback()
416 elif method == 'blockchain.address.subscribe':
418 self.wallet.receive_status_callback(addr, result)
420 elif method == 'blockchain.address.get_history':
422 self.wallet.receive_history_callback(addr, result)
423 self.wallet.was_updated = True
425 elif method == 'blockchain.transaction.broadcast':
426 self.wallet.tx_result = result
427 self.wallet.tx_event.set()
429 elif method == 'blockchain.numblocks.subscribe':
430 self.wallet.blocks = result
431 self.wallet.was_updated = True
433 elif method == 'server.version':
437 print_error("Error: Unknown message:" + method + ", " + params + ", " + result)
440 def start_interface(self):
441 self.interface.start()
442 if self.interface.is_connected:
443 self.wallet.start_session(self.interface)
449 self.start_interface()
451 while self.interface.is_connected:
452 new_addresses = self.wallet.synchronize()
454 self.interface.subscribe(new_addresses)
456 if self.wallet.is_up_to_date():
457 if not self.wallet.up_to_date:
458 self.wallet.up_to_date = True
459 self.wallet.was_updated = True
460 self.wallet.up_to_date_event.set()
462 if self.wallet.up_to_date:
463 self.wallet.up_to_date = False
464 self.wallet.was_updated = True
466 if self.wallet.was_updated:
467 self.wallet.trigger_callbacks()
468 self.wallet.was_updated = False
470 response = self.interface.responses.get()
471 self.handle_response(response)
473 self.wallet.trigger_callbacks()
476 # Server has been changed. Copy callback for new interface.
477 self.proxy = self.interface.proxy
478 self.init_interface()
479 self.start_interface()