X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=909902a0b526f9a8bd8158112e4d4942d0e80ee4;hb=a31101dc21afc4c4f2e32d06141b0ac627396ee7;hp=fccf5d20263f3b2857735e4de35963b45823a78f;hpb=4cbe19b1daad36df75621e558fa6a53eb8d9467b;p=electrum-server.git diff --git a/server.py b/server.py index fccf5d2..909902a 100755 --- a/server.py +++ b/server.py @@ -24,10 +24,12 @@ 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 @@ -65,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 @@ -73,15 +74,9 @@ 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() - - - @@ -110,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): @@ -136,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)) @@ -146,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: @@ -173,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): @@ -216,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) @@ -224,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 @@ -348,6 +334,7 @@ def do_command(cmd, data, ipaddr): session_id, addr = ast.literal_eval(data) except: traceback.print_exc(file=sys.stdout) + print data return None out = add_address_to_session(session_id,addr) @@ -387,130 +374,61 @@ def do_command(cmd, data, ipaddr): return out - -#################################################################### - -def tcp_server_thread(): - thread.start_new_thread(process_input_queue, ()) - thread.start_new_thread(process_output_queue, ()) - - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - s.bind((config.get('server','host'), 50001)) - s.listen(1) +def clean_session_thread(): while not stopping: - conn, addr = s.accept() - try: - thread.start_new_thread(tcp_client_thread, (addr, conn,)) - except: - # can't start new thread if there is no memory.. - traceback.print_exc(file=sys.stdout) - - -def close_session(session_id): - #print "lost connection", session_id - sessions.pop(session_id) - if session_id in sessions_sub_numblocks: - sessions_sub_numblocks.pop(session_id) - - -# one thread per client. put requests in a queue. -def tcp_client_thread(ipaddr,conn): - """ use a persistent connection. put commands in a queue.""" - - print timestr(), "TCP session", ipaddr - global sessions - - session_id = random_string(10) - sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown', 'type':'persistent' } + time.sleep(30) + t = time.time() + for k,s in sessions.items(): + if s.get('type') == 'persistent': continue + t0 = s['last_time'] + if t - t0 > 5*60: + sessions.pop(k) + print "lost session", k + - ipaddr = ipaddr[0] - msg = '' +#################################################################### - while not stopping: - try: - d = conn.recv(1024) - except socket.error: - d = '' - if not d: - close_session(session_id) - break - msg += d - while True: - s = msg.find('\n') - if s ==-1: - break - else: - c = msg[0:s].strip() - msg = msg[s+1:] - if c == 'quit': - conn.close() - close_session(session_id) - return - try: - c = json.loads(c) - except: - print "json error", repr(c) - continue - try: - message_id = c.get('id') - method = c.get('method') - params = c.get('params') - except: - print "syntax error", repr(c), ipaddr - continue - - # add to queue - input_queue.put((session_id, message_id, method, params)) - - - -# read commands from the input queue. perform requests, etc. this should be called from the main thread. -def process_input_queue(): - while not stopping: - session_id, message_id, method, data = input_queue.get() - if session_id not in sessions.keys(): - continue - out = None - if method == 'address.subscribe': - address = data[0] - subscribe_to_address(session_id,message_id,address) - elif method == 'numblocks.subscribe': - subscribe_to_numblocks(session_id,message_id) +import stratum + +class AbeProcessor(stratum.Processor): + def process(self,session,request): + message_id = request['id'] + method = request['method'] + params = request.get('params',[]) + #print request + + result = '' + if method == 'numblocks.subscribe': + session.subscribe_to_numblocks(message_id) + result = block_number + elif method == 'address.subscribe': + address = params[0] + status = store.get_status(address) + session.subscribe_to_address(address,message_id,status) + result = status elif method == 'client.version': - sessions[session_id]['version'] = data[0] + session.version = params[0] elif method == 'server.banner': - out = { 'result':config.get('server','banner').replace('\\n','\n') } + result = config.get('server','banner').replace('\\n','\n') elif method == 'server.peers': - out = { 'result':peer_list.values() } + result = peer_list.values() elif method == 'address.get_history': - address = data[0] - out = { 'result':store.get_history( address ) } + address = params[0] + 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(params[0]) print "sent tx:", txo - out = json.loads(txo) + result = txo else: - print "unknown command", method - if out: - out['id'] = message_id - out = json.dumps( out ) - output_queue.put((session_id, out)) + print "unknown method", request -# this is a separate thread -def process_output_queue(): - while not stopping: - session_id, out = output_queue.get() - session = sessions.get(session_id) - if session: - try: - conn = session.get('conn') - conn.send(out+'\n') - except: - close_session(session_id) - + if result!='': + response = { 'id':message_id, 'result':result } + self.push_response(session,response) + + def get_status(self,addr): + return store.get_status(addr) @@ -519,17 +437,6 @@ def process_output_queue(): -def clean_session_thread(): - while not stopping: - time.sleep(30) - t = time.time() - for k,s in sessions.items(): - if s.get('type') == 'persistent': continue - t0 = s['last_time'] - if t - t0 > 5*60: - sessions.pop(k) - print "lost session", k - def irc_thread(): global peer_list @@ -596,9 +503,6 @@ def http_server_thread(): server.serve_forever() -import traceback - - if __name__ == '__main__': if len(sys.argv)>1: @@ -626,23 +530,31 @@ 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, ()) - thread.start_new_thread(tcp_server_thread, ()) + thread.start_new_thread(http_server_thread, ()) thread.start_new_thread(clean_session_thread, ()) + #tcp stratum + stratum_processor = AbeProcessor() + shared = stratum.Shared() + # Bind shared to processor since constructor is user defined + stratum_processor.shared = shared + stratum_processor.start() + # Create various transports we need + server = stratum.TcpServer(shared, stratum_processor, "ecdsa.org",50001) + server.start() + if (config.get('server','irc') == 'yes' ): thread.start_new_thread(irc_thread, ()) print "starting Electrum server" - + old_block_number = None while not stopping: block_number = store.main_iteration() @@ -650,14 +562,18 @@ if __name__ == '__main__': old_block_number = block_number for session_id in sessions_sub_numblocks.keys(): send_numblocks(session_id) - # do addresses + + stratum_processor.update_from_blocknum(block_number) + while True: try: - addr = address_queue.get(False) + addr = store.address_queue.get(False) except: break do_update_address(addr) + stratum_processor.update_from_address(addr) + time.sleep(10) print "server stopped"