X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=322993d52c16f620ddb782c564869e4eb67d16b8;hb=98a233679a6467cf4473cc239dcda06ca77bba65;hp=56f2b176b7418c1843b4ce5d1aa697a143776037;hpb=72666f93a5b3ff0c0ac3e123f6d8bd938b466479;p=electrum-server.git diff --git a/server.py b/server.py index 56f2b17..322993d 100755 --- a/server.py +++ b/server.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -# Copyright(C) 2011 thomasv@gitorious +# Copyright(C) 2012 thomasv@gitorious # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as @@ -24,19 +24,16 @@ Todo: mempool transactions do not need to be added to the database; it slows it down """ +import abe_backend -import time, json, socket, operator, thread, ast, sys,re + +import time, json, socket, operator, thread, ast, sys, re, traceback import ConfigParser from json import dumps, loads import urllib -# we need to import electrum -sys.path.append('../client/') -from wallet import Wallet -from interface import Interface - config = ConfigParser.ConfigParser() # set some defaults, which will be overwritten by the config file @@ -70,7 +67,6 @@ password = config.get('server','password') stopping = False block_number = -1 -old_block_number = -1 sessions = {} sessions_sub_numblocks = {} # sessions that have subscribed to the service @@ -78,42 +74,10 @@ m_sessions = [{}] # served by http peer_list = {} -wallets = {} # for ultra-light clients such as bccapi - from Queue import Queue input_queue = Queue() output_queue = Queue() -address_queue = Queue() - - - - -class Direct_Interface(Interface): - def __init__(self): - pass - - def handler(self, method, params = ''): - cmds = {'session.new':new_session, - 'session.poll':poll_session, - 'session.update':update_session, - 'transaction.broadcast':send_tx, - 'address.get_history':store.get_history - } - func = cmds[method] - return func( params ) - - - -def send_tx(tx): - postdata = dumps({"method": 'importtransaction', 'params': [tx], 'id':'jsonrpc'}) - respdata = urllib.urlopen(bitcoind_url, postdata).read() - r = loads(respdata) - if r['error'] != None: - out = "error: transaction rejected by memorypool\n"+tx - else: - out = r['result'] - return out @@ -141,23 +105,24 @@ def cmd_load(_,__,pw): -def modified_addresses(session): - if 1: - t1 = time.time() - addresses = session['addresses'] - session['last_time'] = time.time() - ret = {} - k = 0 - for addr in addresses: - status = get_address_status( addr ) - msg_id, last_status = addresses.get( addr ) - if last_status != status: - addresses[addr] = msg_id, status - ret[addr] = status +def modified_addresses(a_session): + #t1 = time.time() + import copy + session = copy.deepcopy(a_session) + addresses = session['addresses'] + session['last_time'] = time.time() + ret = {} + k = 0 + for addr in addresses: + status = store.get_status( addr ) + msg_id, last_status = addresses.get( addr ) + if last_status != status: + addresses[addr] = msg_id, status + ret[addr] = status - t2 = time.time() - t1 - #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 - return ret, addresses + #t2 = time.time() - t1 + #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 + return ret, addresses def poll_session(session_id): @@ -167,6 +132,7 @@ def poll_session(session_id): print time.asctime(), "session not found", session_id return -1, {} else: + sessions[session_id]['last_time'] = time.time() ret, addresses = modified_addresses(session) if ret: sessions[session_id]['addresses'] = addresses return repr( (block_number,ret)) @@ -177,6 +143,7 @@ def poll_session_json(session_id, message_id): if session is None: raise BaseException("session not found %s"%session_id) else: + m_sessions[0][session_id]['last_time'] = time.time() out = [] ret, addresses = modified_addresses(session) if ret: @@ -204,25 +171,13 @@ def do_update_address(addr): addresses = session['addresses'].keys() if addr in addresses: - status = get_address_status( addr ) + status = store.get_status( addr ) message_id, last_status = session['addresses'][addr] if last_status != status: #print "sending new status for %s:"%addr, status send_status(session_id,message_id,addr,status) sessions[session_id]['addresses'][addr] = (message_id,status) -def get_address_status(addr): - # get address status, i.e. the last block for that address. - tx_points = store.get_history(addr) - if not tx_points: - status = None - else: - lastpoint = tx_points[-1] - status = lastpoint['blk_hash'] - # this is a temporary hack; move it up once old clients have disappeared - if status == 'mempool': # and session['version'] != "old": - status = status + ':%d'% len(tx_points) - return status def send_numblocks(session_id): @@ -247,7 +202,7 @@ def subscribe_to_numblocks_json(session_id, message_id): return block_number def subscribe_to_address(session_id, message_id, address): - status = get_address_status(address) + status = store.get_status(address) sessions[session_id]['addresses'][address] = (message_id, status) sessions[session_id]['last_time'] = time.time() send_status(session_id, message_id, address, status) @@ -255,15 +210,15 @@ def subscribe_to_address(session_id, message_id, address): def add_address_to_session_json(session_id, message_id, address): global m_sessions sessions = m_sessions[0] - status = get_address_status(address) + status = store.get_status(address) sessions[session_id]['addresses'][address] = (message_id, status) sessions[session_id]['last_time'] = time.time() m_sessions[0] = sessions return status def add_address_to_session(session_id, address): - status = get_address_status(address) - sessions[session_id]['addresses'][addr] = ("", status) + status = store.get_status(address) + sessions[session_id]['addresses'][address] = ("", status) sessions[session_id]['last_time'] = time.time() return status @@ -378,7 +333,8 @@ def do_command(cmd, data, ipaddr): try: session_id, addr = ast.literal_eval(data) except: - print "error" + traceback.print_exc(file=sys.stdout) + print data return None out = add_address_to_session(session_id,addr) @@ -386,57 +342,10 @@ def do_command(cmd, data, ipaddr): try: session_id, addresses = ast.literal_eval(data) except: - print "error" + traceback.print_exc(file=sys.stdout) return None print timestr(), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses) out = update_session(session_id,addresses) - - elif cmd == 'bccapi_login': - import electrum - print "data",data - v, k = ast.literal_eval(data) - master_public_key = k.decode('hex') # todo: sanitize. no need to decode twice... - print master_public_key - wallet_id = random_string(10) - w = Wallet( Direct_Interface() ) - w.master_public_key = master_public_key.decode('hex') - w.synchronize() - wallets[wallet_id] = w - out = wallet_id - print "wallets", wallets - - elif cmd == 'bccapi_getAccountInfo': - from wallet import int_to_hex - v, wallet_id = ast.literal_eval(data) - w = wallets.get(wallet_id) - if w is not None: - num = len(w.addresses) - c, u = w.get_balance() - out = int_to_hex(num,4) + int_to_hex(c,8) + int_to_hex( c+u, 8 ) - out = out.decode('hex') - else: - print "error",data - out = "error" - - elif cmd == 'bccapi_getAccountStatement': - from wallet import int_to_hex - v, wallet_id = ast.literal_eval(data) - w = wallets.get(wallet_id) - if w is not None: - num = len(w.addresses) - c, u = w.get_balance() - total_records = num_records = 0 - out = int_to_hex(num,4) + int_to_hex(c,8) + int_to_hex( c+u, 8 ) + int_to_hex( total_records ) + int_to_hex( num_records ) - out = out.decode('hex') - else: - print "error",data - out = "error" - - elif cmd == 'bccapi_getSendCoinForm': - out = '' - - elif cmd == 'bccapi_submitTransaction': - out = '' elif cmd=='poll': out = poll_session(data) @@ -447,10 +356,10 @@ def do_command(cmd, data, ipaddr): out = repr( store.get_history( address ) ) elif cmd == 'load': - out = cmd_load(data) + out = cmd_load(None,None,data) elif cmd =='tx': - out = send_tx(data) + out = store.send_tx(data) print timestr(), "sent tx:", ipaddr, out elif cmd == 'stop': @@ -566,10 +475,9 @@ def process_input_queue(): address = data[0] out = { 'result':store.get_history( address ) } elif method == 'transaction.broadcast': - postdata = dumps({"method": 'importtransaction', 'params': [data], 'id':'jsonrpc'}) - txo = urllib.urlopen(bitcoind_url, postdata).read() + txo = store.send_tx(data[0]) print "sent tx:", txo - out = json.loads(txo) + out = {'result':txo } else: print "unknown command", method if out: @@ -664,7 +572,7 @@ def http_server_thread(): server.register_function(cmd_stop, 'stop') server.register_function(cmd_load, 'load') server.register_function(get_banner, 'server.banner') - server.register_function(lambda a,b,c: send_tx(c), 'transaction.broadcast') + server.register_function(lambda a,b,c: store.send_tx(c), 'transaction.broadcast') server.register_function(address_get_history_json, 'address.get_history') server.register_function(add_address_to_session_json, 'address.subscribe') server.register_function(subscribe_to_numblocks_json, 'numblocks.subscribe') @@ -674,9 +582,6 @@ def http_server_thread(): server.serve_forever() -import traceback - - if __name__ == '__main__': if len(sys.argv)>1: @@ -704,11 +609,8 @@ if __name__ == '__main__': print out sys.exit(0) - # backend - import db - store = db.MyStore(config,address_queue) - + store = abe_backend.AbeStore(config) # supported protocols thread.start_new_thread(native_server_thread, ()) @@ -721,7 +623,7 @@ if __name__ == '__main__': print "starting Electrum server" - + old_block_number = None while not stopping: block_number = store.main_iteration() @@ -729,10 +631,9 @@ if __name__ == '__main__': old_block_number = block_number for session_id in sessions_sub_numblocks.keys(): send_numblocks(session_id) - # do addresses while True: try: - addr = address_queue.get(False) + addr = store.address_queue.get(False) except: break do_update_address(addr)