X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=cf1326207892c9e875a95cc0828846fca5c101e8;hb=15aaf022579762f1f8aa584359c2d8db57663858;hp=2d4118a60fe27fd365e3cca979023d7c3334ea57;hpb=bb925d1cf82639af3808483b22ee810144a3f2f9;p=electrum-server.git

diff --git a/server.py b/server.py
index 2d4118a..cf13262 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')
@@ -77,7 +76,9 @@ stopping = False
 block_number = -1
 old_block_number = -1
 sessions = {}
-sessions_sub_numblocks = [] # sessions that have subscribed to the service
+sessions_sub_numblocks = {} # sessions that have subscribed to the service
+
+m_sessions = [{}] # served by http
 
 dblock = thread.allocate_lock()
 peer_list = {}
@@ -89,11 +90,24 @@ 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 "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)
@@ -227,10 +241,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:
@@ -334,7 +347,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
@@ -349,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 )
@@ -375,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()
@@ -418,16 +429,47 @@ 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:
+        print "poll: session found", session_id
+        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
 
@@ -435,22 +477,19 @@ 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:
-            print "address ", addr, "found in session", session_id
             status = get_address_status( addr )
-            print "new_status:", status
-            last_status = session['addresses'][addr]
-            print "last_status", last_status
+            message_id, last_status = session['addresses'][addr]
             if last_status != status:
-                print "status is new", addr
-                send_status(session_id,addr,status)
-                sessions[session_id]['addresses'][addr] = 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.
@@ -467,34 +506,79 @@ def get_address_status(addr):
 
 
 def send_numblocks(session_id):
-    out = json.dumps( {'method':'numblocks.subscribe', 'result':block_number} )
+    message_id = sessions_sub_numblocks[session_id]
+    out = json.dumps( {'id':message_id, '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 } )
+def send_status(session_id, message_id, address, status):
+    out = json.dumps( { 'id':message_id, 'result':status } )
     output_queue.put((session_id, out))
 
-def subscribe_to_numblocks(session_id):
-    sessions_sub_numblocks.append(session_id)
+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_address(session_id, address):
+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]['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]['type'] = 'subscribe'
-    sessions[session_id]['addresses'][address] = status
+    sessions[session_id]['addresses'][addr] = ("", status)
     sessions[session_id]['last_time'] = time.time()
-    send_status(session_id, address, status)
+    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] = ''
@@ -547,12 +631,12 @@ def native_client_thread(ipaddr,conn):
         conn.close()
 
 
+def timestr():
+    return time.strftime("[%d/%m/%Y-%H:%M:%S]")
 
 # used by the native handler
 def do_command(cmd, data, ipaddr):
 
-    timestr = time.strftime("[%d/%m/%Y-%H:%M:%S]")
-
     if cmd=='b':
         out = "%d"%block_number
 
@@ -567,16 +651,24 @@ def do_command(cmd, data, ipaddr):
         except:
             print "error", data
             return None
-        print timestr, "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
+        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)
         except:
             print "error"
             return None
-        print timestr, "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
+        print timestr(), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
         out = update_session(session_id,addresses)
 
     elif cmd == 'bccapi_login':
@@ -639,7 +731,7 @@ def do_command(cmd, data, ipaddr):
 
     elif cmd =='tx':
         out = send_tx(data)
-        print timestr, "sent tx:", ipaddr, out
+        print timestr(), "sent tx:", ipaddr, out
 
     elif cmd == 'stop':
         out = cmd_stop(data)
@@ -673,73 +765,96 @@ def tcp_server_thread():
             traceback.print_exc(file=sys.stdout)
 
 
-def close_sesion(session_id):
-    print "lost connection", session_id
+def close_session(session_id):
+    #print "lost connection", session_id
     sessions.pop(session_id)
-    sessions_sub_numblocks.remove(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 "persistent client thread", ipaddr
+
+    print timestr(), "TCP session", ipaddr
     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 = ''
 
     while not stopping:
-        d = conn.recv(1024)
-        msg += d
+        try:
+            d = conn.recv(1024)
+        except socket.error:
+            d = ''
         if not d:
-            close_sesion(session_id)
+            close_session(session_id)
             break
 
+        msg += d
         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:
+                    c = json.loads(c)
+                except:
+                    print "json error", repr(c)
+                    continue
                 try:
-                    cmd = c['method']
-                    data = c['params']
+                    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, cmd, data))
+                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, cmd, data = input_queue.get()
+        session_id, message_id, method, data = input_queue.get()
+        if session_id not in sessions.keys():
+            continue
         out = None
-        if cmd == 'address.subscribe':
-            subscribe_to_address(session_id,data)
-        elif cmd == 'numblocks.subscribe':
-            subscribe_to_numblocks(session_id)
-        elif cmd == 'client.version':
-            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 == 'address.get_history':
-            address = data
-            out = json.dumps( { 'method':'address.get_history', 'address':address, 'result':store.get_history( address ) } )
-        elif cmd == 'transaction.broadcast':
-            txo = send_tx(data)
+        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)
+        elif method == 'client.version':
+            sessions[session_id]['version'] = data[0]
+        elif method == 'server.banner':
+            out = { 'result':config.get('server','banner').replace('\\n','\n') } 
+        elif method == 'server.peers':
+            out = { 'result':peer_list.values() } 
+        elif method == 'address.get_history':
+            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()
             print "sent tx:", txo
-            out = json.dumps( { 'method':'transaction.broadcast', 'result':txo } )
+            out = json.loads(txo)
         else:
-            print "unknown command", cmd
+            print "unknown command", method
         if out:
+            out['id'] = message_id
+            out = json.dumps( out )
             output_queue.put((session_id, out))
 
 # this is a separate thread
@@ -761,7 +876,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'})
@@ -776,23 +893,24 @@ 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
+            if s.get('type') == 'persistent': continue
             t0 = s['last_time']
             if t - t0 > 5*60:
                 sessions.pop(k)
@@ -841,24 +959,29 @@ 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(send_tx, '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(create_session_json, 'session.create') #internal message
+    server.register_function(poll_session_json, 'session.poll')
+    server.register_function(subscribe_to_numblocks_json, 'numblocks.subscribe')
+    server.register_function(client_version_json, 'client.version')
+    server.register_function(lambda a,b:None, 'ping')
     server.serve_forever()
 
 
@@ -874,7 +997,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':
@@ -882,11 +1005,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
@@ -894,7 +1017,6 @@ if __name__ == '__main__':
 
 
     print "starting Electrum server"
-    print "cache:", config.get('server', 'cache')
 
     conf = DataStore.CONFIG_DEFAULTS
     args, argv = readconf.parse_argv( [], conf)
@@ -924,11 +1046,11 @@ 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:
+                for session_id in sessions_sub_numblocks.keys():
                     send_numblocks(session_id)
 
         except IOError: