X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=8050c55e413eabd9441cab30ae0e1a49b42d4338;hb=1ec6053579c72699825d723bb1ef41c702e8c428;hp=28851e9c7affe0998e926e6827a3b7c6c13fa6c6;hpb=9bd31ae590b10ab50d0b6fd1f3f59fe2e1a3e7f9;p=electrum-server.git diff --git a/server.py b/server.py index 28851e9..8050c55 100755 --- a/server.py +++ b/server.py @@ -50,7 +50,6 @@ config.set('server', 'host', 'localhost') config.set('server', 'port', 50000) config.set('server', 'password', '') config.set('server', 'irc', 'yes') -config.set('server', 'cache', 'no') config.set('server', 'ircname', 'Electrum server') config.add_section('database') config.set('database', 'type', 'psycopg2') @@ -87,12 +86,24 @@ wallets = {} # for ultra-light clients such as bccapi from Queue import Queue input_queue = Queue() output_queue = Queue() +address_queue = Queue() class MyStore(Datastore_class): - def import_tx(self, tx, is_coinbase): - tx_id = super(MyStore, self).import_tx(tx, is_coinbase) - if config.get('server', 'cache') == 'yes': self.update_tx_cache(tx_id) + def import_block(self, b, chain_ids=frozenset()): + block_id = super(MyStore, self).import_block(b, chain_ids) + print "import block", block_id + for pos in xrange(len(b['transactions'])): + tx = b['transactions'][pos] + if 'hash' not in tx: + tx['hash'] = util.double_sha256(tx['tx']) + tx_id = store.tx_find_id_and_value(tx) + if tx_id: + self.update_tx_cache(tx_id) + else: + print "error: import_block: no tx_id" + return block_id + def update_tx_cache(self, txid): inrows = self.get_tx_inputs(txid, False) @@ -102,6 +113,8 @@ class MyStore(Datastore_class): if self.tx_cache.has_key(address): print "cache: invalidating", address self.tx_cache.pop(address) + address_queue.put(address) + outrows = self.get_tx_outputs(txid, False) for row in outrows: _hash = store.binout(row[6]) @@ -109,6 +122,7 @@ class MyStore(Datastore_class): if self.tx_cache.has_key(address): print "cache: invalidating", address self.tx_cache.pop(address) + address_queue.put(address) def safe_sql(self,sql, params=(), lock=True): try: @@ -223,10 +237,9 @@ class MyStore(Datastore_class): def get_history(self, addr): - if config.get('server','cache') == 'yes': - cached_version = self.tx_cache.get( addr ) - if cached_version is not None: - return cached_version + cached_version = self.tx_cache.get( addr ) + if cached_version is not None: + return cached_version version, binaddr = decode_check_address(addr) if binaddr is None: @@ -249,7 +262,6 @@ class MyStore(Datastore_class): tx_hash = self.hashout_hex(tx_hash) txpoint = { "nTime": int(nTime), - #"chain_id": int(chain_id), "height": int(height), "is_in": int(is_in), "blk_hash": self.hashout_hex(blk_hash), @@ -288,7 +300,6 @@ class MyStore(Datastore_class): #print "mempool", tx_hash txpoint = { "nTime": 0, - #"chain_id": 1, "height": 0, "is_in": int(is_in), "blk_hash": 'mempool', @@ -332,7 +343,7 @@ class MyStore(Datastore_class): if not row[4]: txpoint['raw_scriptPubKey'] = row[1] # cache result - if config.get('server','cache') == 'yes' and not address_has_mempool: + if not address_has_mempool: self.tx_cache[addr] = txpoints return txpoints @@ -429,8 +440,27 @@ def poll_session(session_id): return out + +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 + addresses = session['addresses'].keys() + + if addr in addresses: + print "address ", addr, "is watched by", session_id + status = get_address_status( addr ) + last_status = session['addresses'][addr] + if last_status != status: + print "sending new status for %s:"%addr, status + send_status(session_id,addr,status) + sessions[session_id]['addresses'][addr] = status + + def get_address_status(addr): - # get addtess status, i.e. the last block for that address. + # get address status, i.e. the last block for that address. tx_points = store.get_history(addr) if not tx_points: status = None @@ -447,18 +477,20 @@ def send_numblocks(session_id): out = json.dumps( {'method':'numblocks.subscribe', 'result':block_number} ) output_queue.put((session_id, out)) +def send_status(session_id, address, status): + out = json.dumps( { 'method':'address.subscribe', 'address':address, 'status':status } ) + output_queue.put((session_id, out)) + def subscribe_to_numblocks(session_id): sessions_sub_numblocks.append(session_id) send_numblocks(session_id) def subscribe_to_address(session_id, address): - #print "%s subscribing to %s"%(session_id,address) status = get_address_status(address) sessions[session_id]['type'] = 'subscribe' sessions[session_id]['addresses'][address] = status sessions[session_id]['last_time'] = time.time() - out = json.dumps( { 'method':'address.subscribe', 'address':address, 'status':status } ) - output_queue.put((session_id, out)) + send_status(session_id, address, status) def new_session(version, addresses): session_id = random_string(10) @@ -648,6 +680,13 @@ def tcp_server_thread(): 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.remove(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.""" @@ -661,20 +700,34 @@ def tcp_client_thread(ipaddr,conn): msg = '' while not stopping: - d = conn.recv(1024) + try: + d = conn.recv(1024) + except socket.error: + d = '' + if not d: + close_session(session_id) + break + msg += d - if not d: break while True: s = msg.find('\n') if s ==-1: break else: - c = msg[0:s] + c = msg[0:s].strip() msg = msg[s+1:] - c = json.loads(c) + if c == 'quit': + conn.close() + close_session(session_id) + return try: - cmd = c['method'] - data = c['params'] + c = json.loads(c) + except: + print "json error", repr(c) + continue + try: + cmd = c.get('method') + data = c.get('params') except: print "syntax error", repr(c), ipaddr continue @@ -683,10 +736,13 @@ def tcp_client_thread(ipaddr,conn): input_queue.put((session_id, cmd, data)) + # 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, cmd, data = input_queue.get() + if session_id not in sessions.keys(): + continue out = None if cmd == 'address.subscribe': subscribe_to_address(session_id,data) @@ -696,11 +752,15 @@ def process_input_queue(): sessions[session_id]['version'] = data elif cmd == 'server.banner': out = json.dumps( { 'method':'server.banner', 'result':config.get('server','banner').replace('\\n','\n') } ) + elif cmd == 'server.peers': + out = json.dumps( { 'method':'server.peers', 'result':peer_list.values() } ) elif cmd == 'address.get_history': address = data out = json.dumps( { 'method':'address.get_history', 'address':address, 'result':store.get_history( address ) } ) elif cmd == 'transaction.broadcast': - out = json.dumps( { 'method':'transaction.broadcast', 'result':send_tx(data) } ) + txo = send_tx(data) + print "sent tx:", txo + out = json.dumps( { 'method':'transaction.broadcast', 'result':txo } ) else: print "unknown command", cmd if out: @@ -712,8 +772,12 @@ def process_output_queue(): session_id, out = output_queue.get() session = sessions.get(session_id) if session: - conn = session.get('conn') - conn.send(out+'\n') + try: + conn = session.get('conn') + conn.send(out+'\n') + except: + close_session(session_id) + @@ -721,7 +785,9 @@ def process_output_queue(): def memorypool_update(store): + ds = BCDataStream.BCDataStream() + previous_transactions = store.mempool_keys store.mempool_keys = [] postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'}) @@ -736,25 +802,28 @@ def memorypool_update(store): ds.write(hextx.decode('hex')) tx = deserialize.parse_Transaction(ds) tx['hash'] = util.double_sha256(tx['tx']) - tx_hash = tx['hash'][::-1].encode('hex') + tx_hash = store.hashin(tx['hash']) + store.mempool_keys.append(tx_hash) if store.tx_find_id_and_value(tx): pass else: - store.import_tx(tx, False) + tx_id = store.import_tx(tx, False) + store.update_tx_cache(tx_id) store.commit() - def clean_session_thread(): while not stopping: time.sleep(30) t = time.time() for k,s in sessions.items(): + if s.get('type') == 'subscribe': continue t0 = s['last_time'] if t - t0 > 5*60: sessions.pop(k) + print "lost session", k def irc_thread(): @@ -852,7 +921,6 @@ if __name__ == '__main__': print "starting Electrum server" - print "cache:", config.get('server', 'cache') conf = DataStore.CONFIG_DEFAULTS args, argv = readconf.parse_argv( [], conf) @@ -873,6 +941,7 @@ if __name__ == '__main__': thread.start_new_thread(http_server_thread, (store,)) thread.start_new_thread(clean_session_thread, ()) + if (config.get('server','irc') == 'yes' ): thread.start_new_thread(irc_thread, ()) @@ -881,8 +950,8 @@ if __name__ == '__main__': dblock.acquire() store.catch_up() memorypool_update(store) - block_number = store.get_block_number(1) + block_number = store.get_block_number(1) if block_number != old_block_number: old_block_number = block_number for session_id in sessions_sub_numblocks: @@ -896,6 +965,15 @@ if __name__ == '__main__': block_number = 0 finally: dblock.release() + + # do addresses + while True: + try: + addr = address_queue.get(False) + except: + break + do_update_address(addr) + time.sleep(10) print "server stopped"