X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=server.py;h=42b6b9f487a3cf52384b955f21ab69517995ca9b;hb=d9bbde33c1faca005c47bce650650907991d9c51;hp=5802a8d349e5d34f74dc55c8bed4908c431f7280;hpb=21b133f4706a55f317e322a6b2e593b5a0750707;p=electrum-server.git

diff --git a/server.py b/server.py
index 5802a8d..42b6b9f 100755
--- a/server.py
+++ b/server.py
@@ -75,12 +75,20 @@ bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.g
 
 stopping = False
 block_number = -1
+old_block_number = -1
 sessions = {}
+sessions_sub_numblocks = [] # sessions that have subscribed to the service
+
 dblock = thread.allocate_lock()
 peer_list = {}
 
 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):
@@ -95,6 +103,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])
@@ -102,6 +112,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:
@@ -217,8 +228,8 @@ 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: 
+            cached_version = self.tx_cache.get( addr )
+            if cached_version is not None:
                 return cached_version
 
         version, binaddr = decode_check_address(addr)
@@ -242,7 +253,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),
@@ -281,7 +291,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', 
@@ -408,18 +417,7 @@ def poll_session(session_id):
         k = 0
         for addr in addresses:
             if store.tx_cache.get( addr ) is not None: k += 1
-
-            # get addtess 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)
-
+            status = get_address_status( addr )
             last_status = addresses.get( addr )
             if last_status != status:
                 addresses[addr] = status
@@ -434,6 +432,59 @@ 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, "found in session", session_id
+            status = get_address_status( addr )
+            print "new_status:", status
+            last_status = session['addresses'][addr]
+            print "last_status", last_status
+            if last_status != status:
+                print "status is new", addr
+                send_status(session_id,addr,status)
+                sessions[session_id]['addresses'][addr] = 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):
+    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):
+    status = get_address_status(address)
+    sessions[session_id]['type'] = 'subscribe'
+    sessions[session_id]['addresses'][address] = status
+    sessions[session_id]['last_time'] = time.time()
+    send_status(session_id, address, status)
+
 def new_session(version, addresses):
     session_id = random_string(10)
     sessions[session_id] = { 'addresses':{}, 'version':version }
@@ -443,11 +494,6 @@ def new_session(version, addresses):
     sessions[session_id]['last_time'] = time.time()
     return out
 
-def subscribe_to_address(session_id, address):
-    sessions[session_id]['addresses'][address] = ''
-    sessions[session_id]['last_time'] = time.time()
-
-
 def update_session(session_id,addresses):
     sessions[session_id]['addresses'] = {}
     for a in addresses:
@@ -611,6 +657,9 @@ def do_command(cmd, data, ipaddr):
 ####################################################################
 
 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))
@@ -624,22 +673,32 @@ 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."""
     print "persistent client thread", ipaddr
     global sessions
 
     session_id = random_string(10)
-    sessions[session_id] = { 'addresses':{}, 'version':'unknown' }
+    sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown' }
 
     ipaddr = ipaddr[0]
     msg = ''
 
-    while True:
+    while not stopping:
         d = conn.recv(1024)
         msg += d
-        if not d: break
+        if not d:
+            close_session(session_id)
+            break
+
         while True:
             s = msg.find('\n')
             if s ==-1:
@@ -655,18 +714,51 @@ def tcp_client_thread(ipaddr,conn):
                     print "syntax error", repr(c), ipaddr
                     continue
 
-                out = None
-                if cmd == 'blockchain.address.subscribe':
-                    subscribe_to_address(session_id,data)
-                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') } )
-                else:
-                    print "unknown command", cmd
+                # add to queue
+                input_queue.put((session_id, cmd, data))
+
 
-                if out:
-                    conn.send(out+'\n')
+# 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)
+        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 == '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':
+            txo = send_tx(data)
+            print "sent tx:", txo
+            out = json.dumps( { 'method':'transaction.broadcast', 'result':txo } )
+        else:
+            print "unknown command", cmd
+        if out:
+            output_queue.put((session_id, out))
+
+# 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)
+                
 
 
 
@@ -705,9 +797,11 @@ def clean_session_thread():
         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():
@@ -826,6 +920,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, ())
 
@@ -835,6 +930,12 @@ if __name__ == '__main__':
             store.catch_up()
             memorypool_update(store)
             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:
+                    send_numblocks(session_id)
+
         except IOError:
             print "IOError: cannot reach bitcoind"
             block_number = 0
@@ -843,6 +944,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"