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
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'")
38 class Interface(threading.Thread):
39 def __init__(self, host, port):
40 threading.Thread.__init__(self)
45 self.servers = [] # actual list from IRC
47 self.bytes_received = 0
49 self.is_connected = True
50 self.poll_interval = 1
54 self.responses = Queue.Queue()
55 self.unanswered_requests = {}
57 def init_socket(self):
61 # push a fake response so that the getting thread exits its loop
62 self.responses.put(None)
64 def queue_json_response(self, c):
68 error = c.get('error')
71 print "received error:", c
74 if msg_id is not None:
75 method, params = self.unanswered_requests.pop(msg_id)
76 result = c.get('result')
79 method = c.get('method')
80 params = c.get('params')
82 if method == 'blockchain.numblocks.subscribe':
86 elif method == 'blockchain.address.subscribe':
91 self.responses.put({'method':method, 'params':params, 'result':result})
95 def subscribe(self, addresses):
97 for addr in addresses:
98 messages.append(('blockchain.address.subscribe', [addr]))
104 class PollingInterface(Interface):
105 """ non-persistent connection. synchronous calls"""
107 def __init__(self, host, port):
108 Interface.__init__(self, host, port)
109 self.session_id = None
111 def get_history(self, address):
112 self.send([('blockchain.address.get_history', [address] )])
116 #if is_new or wallet.remote_url:
117 # self.was_updated = True
118 # is_new = wallet.synchronize()
119 # wallet.update_tx_history()
126 self.is_connected = True
127 while self.is_connected:
131 time.sleep(self.poll_interval)
132 except socket.gaierror:
137 traceback.print_exc(file=sys.stdout)
140 self.is_connected = False
151 class HttpStratumInterface(PollingInterface):
156 def send(self, messages):
157 import urllib2, json, time, cookielib
159 cj = cookielib.CookieJar()
160 opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
161 urllib2.install_opener(opener)
168 if type(params) != type([]): params = [params]
169 data.append( { 'method':method, 'id':self.message_id, 'params':params } )
170 self.unanswered_requests[self.message_id] = method, params
174 data_json = json.dumps(data)
179 host = 'http://%s:%d'%( self.host, self.port )
180 headers = {'content-type': 'application/json'}
182 headers['cookie'] = 'SESSION=%s'%self.session_id
184 req = urllib2.Request(host, data_json, headers)
185 response_stream = urllib2.urlopen(req)
187 for index, cookie in enumerate(cj):
188 if cookie.name=='SESSION':
189 self.session_id = cookie.value
191 response = response_stream.read()
192 self.bytes_received += len(response)
194 response = json.loads( response )
195 if type(response) is not type([]):
196 self.queue_json_response(response)
198 for item in response:
199 self.queue_json_response(item)
202 self.poll_interval = 1
204 if self.poll_interval < 15:
205 self.poll_interval += 1
206 #print self.poll_interval, response
208 self.rtime = time.time() - t1
209 self.is_connected = True
214 class TcpStratumInterface(Interface):
215 """json-rpc over persistent TCP connection, asynchronous"""
217 def __init__(self, host, port):
218 Interface.__init__(self, host, port)
220 def init_socket(self):
221 self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
222 self.s.settimeout(60)
223 self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
225 self.s.connect(( self.host, self.port))
226 self.is_connected = True
227 self.send([('server.version', [ELECTRUM_VERSION])])
228 print "Connected to %s:%d"%(self.host,self.port)
230 self.is_connected = False
231 print "Not connected"
236 while self.is_connected:
237 try: msg = self.s.recv(1024)
238 except socket.timeout:
239 # ping the server with server.version, as a real ping does not exist yet
240 self.send([('server.version', [ELECTRUM_VERSION])])
243 self.bytes_received += len(msg)
245 self.is_connected = False
246 print "disconnected."
254 self.queue_json_response(c)
257 traceback.print_exc(file=sys.stdout)
259 self.is_connected = False
263 def send(self, messages):
267 request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
268 self.unanswered_requests[self.message_id] = method, params
271 out += request + '\n'
274 sent = self.s.send( out )
277 def get_history(self, addr):
278 self.send([('blockchain.address.get_history', [addr])])
284 class WalletSynchronizer(threading.Thread):
286 def __init__(self, wallet, loop=False):
287 threading.Thread.__init__(self)
291 self.init_interface()
293 def init_interface(self):
295 host, port, protocol = self.wallet.server.split(':')
298 self.wallet.pick_random_server()
299 host, port, protocol = self.wallet.server.split(':')
302 #print protocol, host, port
304 InterfaceClass = TcpStratumInterface
305 elif protocol == 'h':
306 InterfaceClass = HttpStratumInterface
308 print "unknown protocol"
309 InterfaceClass = TcpStratumInterface
311 self.interface = InterfaceClass(host, port)
312 self.wallet.interface = self.interface
315 def handle_response(self, r):
323 if method == 'server.banner':
324 self.wallet.banner = result
325 self.wallet.was_updated = True
327 elif method == 'server.peers.subscribe':
335 if re.match("[th]\d+",v):
336 ports.append((v[0],v[1:]))
338 servers.append( (host, ports) )
339 self.interface.servers = servers
341 elif method == 'blockchain.address.subscribe':
343 self.wallet.receive_status_callback(addr, result)
345 elif method == 'blockchain.address.get_history':
347 self.wallet.receive_history_callback(addr, result)
348 self.wallet.was_updated = True
350 elif method == 'blockchain.transaction.broadcast':
351 self.wallet.tx_result = result
352 self.wallet.tx_event.set()
354 elif method == 'blockchain.numblocks.subscribe':
355 self.wallet.blocks = result
357 elif method == 'server.version':
361 print "unknown message:", method, params, result
364 def start_interface(self):
365 self.interface.init_socket()
366 self.interface.start()
367 if self.interface.is_connected:
368 self.wallet.start_session(self.interface)
374 self.start_interface()
376 while self.interface.is_connected:
377 new_addresses = self.wallet.synchronize()
379 self.interface.subscribe(new_addresses)
381 if self.wallet.is_up_to_date():
382 if not self.wallet.up_to_date:
383 self.wallet.up_to_date = True
384 self.wallet.was_updated = True
385 self.wallet.up_to_date_event.set()
387 if self.wallet.up_to_date:
388 self.wallet.up_to_date = False
389 self.wallet.was_updated = True
391 if self.wallet.was_updated:
392 self.wallet.gui_callback()
393 self.wallet.was_updated = False
395 response = self.interface.responses.get()
396 self.handle_response(response)
398 self.wallet.gui_callback()
401 self.init_interface()
402 self.start_interface()