X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=d8a3683a502f9f99520876667cd899a0eb8331f0;hb=851dc213489be9439a841ee168f0029bae281e6d;hp=9caf3530ef277f13c6ef64618ab193952a381f1c;hpb=45062a27b514a45237d640ec7de133c98d3e475b;p=electrum-server.git diff --git a/server.py b/server.py index 9caf353..d8a3683 100755 --- a/server.py +++ b/server.py @@ -78,6 +78,8 @@ old_block_number = -1 sessions = {} sessions_sub_numblocks = {} # sessions that have subscribed to the service +m_sessions = [{}] # served by http + dblock = thread.allocate_lock() peer_list = {} @@ -88,6 +90,8 @@ input_queue = Queue() output_queue = Queue() address_queue = Queue() + + class MyStore(Datastore_class): def import_block(self, b, chain_ids=frozenset()): @@ -358,8 +362,8 @@ class Direct_Interface(Interface): cmds = {'session.new':new_session, 'session.poll':poll_session, 'session.update':update_session, - 'blockchain.transaction.broadcast':send_tx, - 'blockchain.address.get_history':store.get_history + 'transaction.broadcast':send_tx, + 'address.get_history':store.get_history } func = cmds[method] return func( params ) @@ -384,41 +388,39 @@ def random_string(N): -def cmd_stop(data): +def cmd_stop(_,__,pw): global stopping - if password == data: + if password == pw: stopping = True return 'ok' else: return 'wrong password' -def cmd_load(pw): +def cmd_load(_,__,pw): if password == pw: return repr( len(sessions) ) else: return 'wrong password' -def clear_cache(pw): +def clear_cache(_,__,pw): if password == pw: store.tx_cache = {} return 'ok' else: return 'wrong password' -def get_cache(pw,addr): +def get_cache(_,__,pw,addr): if password == pw: return store.tx_cache.get(addr) else: return 'wrong password' -def poll_session(session_id): - session = sessions.get(session_id) - if session is None: - print time.asctime(), "session not found", session_id - out = repr( (-1, {})) - else: + + +def modified_addresses(session): + if 1: t1 = time.time() addresses = session['addresses'] session['last_time'] = time.time() @@ -427,16 +429,46 @@ def poll_session(session_id): for addr in addresses: if store.tx_cache.get( addr ) is not None: k += 1 status = get_address_status( addr ) - last_status = addresses.get( addr ) + msg_id, last_status = addresses.get( addr ) if last_status != status: - addresses[addr] = status + addresses[addr] = msg_id, status ret[addr] = status - if ret: - sessions[session_id]['addresses'] = addresses - out = repr( (block_number, ret ) ) + t2 = time.time() - t1 - if t2 > 10: - print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 + #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2 + return ret, addresses + + +def poll_session(session_id): + # native + session = sessions.get(session_id) + if session is None: + print time.asctime(), "session not found", session_id + return -1, {} + else: + ret, addresses = modified_addresses(session) + if ret: sessions[session_id]['addresses'] = addresses + return repr( (block_number,ret)) + + +def poll_session_json(session_id, message_id): + session = m_sessions[0].get(session_id) + if session is None: + raise BaseException("session not found %s"%session_id) + else: + out = [] + ret, addresses = modified_addresses(session) + if ret: + m_sessions[0][session_id]['addresses'] = addresses + for addr in ret: + msg_id, status = addresses[addr] + out.append( { 'id':msg_id, 'result':status } ) + + msg_id, last_nb = session.get('numblocks') + if last_nb: + if last_nb != block_number: + m_sessions[0][session_id]['numblocks'] = msg_id, block_number + out.append( {'id':msg_id, 'result':block_number} ) return out @@ -444,9 +476,10 @@ def poll_session(session_id): def do_update_address(addr): # an address was involved in a transaction; we check if it was subscribed to in a session # the address can be subscribed in several sessions; the cache should ensure that we don't do redundant requests + for session_id in sessions.keys(): session = sessions[session_id] - if session.get('type') != 'subscribe': continue + if session.get('type') != 'persistent': continue addresses = session['addresses'].keys() if addr in addresses: @@ -457,7 +490,6 @@ def do_update_address(addr): 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) @@ -481,27 +513,71 @@ def send_status(session_id, message_id, address, status): out = json.dumps( { 'id':message_id, 'result':status } ) output_queue.put((session_id, out)) +def address_get_history_json(_,message_id,address): + return store.get_history(address) + def subscribe_to_numblocks(session_id, message_id): sessions_sub_numblocks[session_id] = message_id send_numblocks(session_id) +def subscribe_to_numblocks_json(session_id, message_id): + global m_sessions + m_sessions[0][session_id]['numblocks'] = message_id,block_number + return block_number + def subscribe_to_address(session_id, message_id, address): status = get_address_status(address) - sessions[session_id]['type'] = 'subscribe' sessions[session_id]['addresses'][address] = (message_id, status) sessions[session_id]['last_time'] = time.time() send_status(session_id, message_id, address, status) +def add_address_to_session_json(session_id, message_id, address): + global m_sessions + sessions = m_sessions[0] + status = get_address_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) + sessions[session_id]['last_time'] = time.time() + return status + def new_session(version, addresses): session_id = random_string(10) sessions[session_id] = { 'addresses':{}, 'version':version } for a in addresses: - sessions[session_id]['addresses'][a] = '' + sessions[session_id]['addresses'][a] = ('','') out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) ) sessions[session_id]['last_time'] = time.time() return out + +def client_version_json(session_id, _, version): + global m_sessions + sessions = m_sessions[0] + sessions[session_id]['version'] = version + m_sessions[0] = sessions + +def create_session_json(_, __): + sessions = m_sessions[0] + session_id = random_string(10) + print "creating session", session_id + sessions[session_id] = { 'addresses':{}, 'numblocks':('','') } + sessions[session_id]['last_time'] = time.time() + m_sessions[0] = sessions + return session_id + + + +def get_banner(_,__): + return config.get('server','banner').replace('\\n','\n') + def update_session(session_id,addresses): + """deprecated in 0.42""" sessions[session_id]['addresses'] = {} for a in addresses: sessions[session_id]['addresses'][a] = '' @@ -577,6 +653,14 @@ def do_command(cmd, data, ipaddr): print timestr(), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version out = new_session(version, addresses) + elif cmd=='address.subscribe': + try: + session_id, addr = ast.literal_eval(data) + except: + print "error" + return None + out = add_address_to_session(session_id,addr) + elif cmd=='update_session': try: session_id, addresses = ast.literal_eval(data) @@ -681,7 +765,7 @@ def tcp_server_thread(): def close_session(session_id): - print "lost connection", session_id + #print "lost connection", session_id sessions.pop(session_id) if session_id in sessions_sub_numblocks: sessions_sub_numblocks.pop(session_id) @@ -695,7 +779,7 @@ def tcp_client_thread(ipaddr,conn): global sessions session_id = random_string(10) - sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown' } + sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown', 'type':'persistent' } ipaddr = ipaddr[0] msg = '' @@ -761,9 +845,10 @@ def process_input_queue(): address = data[0] out = { 'result':store.get_history( address ) } elif method == 'transaction.broadcast': - txo = send_tx(data) + postdata = dumps({"method": 'importtransaction', 'params': [data], 'id':'jsonrpc'}) + txo = urllib.urlopen(bitcoind_url, postdata).read() print "sent tx:", txo - out = { 'result':txo } + out = json.loads(txo) else: print "unknown command", method if out: @@ -824,7 +909,7 @@ def clean_session_thread(): time.sleep(30) t = time.time() for k,s in sessions.items(): - if s.get('type') == 'subscribe': continue + if s.get('type') == 'persistent': continue t0 = s['last_time'] if t - t0 > 5*60: sessions.pop(k) @@ -873,24 +958,28 @@ def irc_thread(): s.close() +def get_peers_json(_,__): + return peer_list.values() def http_server_thread(store): # see http://code.google.com/p/jsonrpclib/ from SocketServer import ThreadingMixIn - from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer - class SimpleThreadedJSONRPCServer(ThreadingMixIn, SimpleJSONRPCServer): pass - server = SimpleThreadedJSONRPCServer(( config.get('server','host'), 8081)) - server.register_function(lambda : peer_list.values(), 'peers') + from StratumJSONRPCServer import StratumJSONRPCServer + class StratumThreadedJSONRPCServer(ThreadingMixIn, StratumJSONRPCServer): pass + server = StratumThreadedJSONRPCServer(( config.get('server','host'), 8081)) + server.register_function(get_peers_json, 'server.peers') server.register_function(cmd_stop, 'stop') server.register_function(cmd_load, 'load') - server.register_function(lambda : block_number, 'blocks') server.register_function(clear_cache, 'clear_cache') server.register_function(get_cache, 'get_cache') - server.register_function(send_tx, 'blockchain.transaction.broadcast') - server.register_function(store.get_history, 'blockchain.address.get_history') - server.register_function(new_session, 'session.new') - server.register_function(update_session, 'session.update') - server.register_function(poll_session, 'session.poll') + server.register_function(get_banner, 'server.banner') + server.register_function(lambda a,b,c: 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') + server.register_function(client_version_json, 'client.version') + server.register_function(create_session_json, 'session.create') # internal message (not part of protocol) + server.register_function(poll_session_json, 'session.poll') # internal message (not part of protocol) server.serve_forever() @@ -906,7 +995,7 @@ if __name__ == '__main__': if cmd == 'load': out = server.load(password) elif cmd == 'peers': - out = server.peers() + out = server.server.peers() elif cmd == 'stop': out = server.stop(password) elif cmd == 'clear_cache': @@ -914,11 +1003,11 @@ if __name__ == '__main__': elif cmd == 'get_cache': out = server.get_cache(password,sys.argv[2]) elif cmd == 'h': - out = server.blockchain.address.get_history(sys.argv[2]) + out = server.address.get_history(sys.argv[2]) elif cmd == 'tx': - out = server.blockchain.transaction.broadcast(sys.argv[2]) + out = server.transaction.broadcast(sys.argv[2]) elif cmd == 'b': - out = server.blocks() + out = server.numblocks.subscribe() else: out = "Unknown command: '%s'" % cmd print out