X-Git-Url: https://git.novaco.in/?p=electrum-server.git;a=blobdiff_plain;f=backends%2Fbitcoind%2Fblockchain_processor.py;h=3b65b27953498e08a3ec1a7f887632e4fb6b9ca2;hp=d4c56f4474630e2971ea4f1e4a031c01c7292097;hb=4ce69b7ea24ead59ebbcc7ed335ea9762ae3724b;hpb=aa444a8b7c461ca2df014f015dc64c6226fd976e

diff --git a/backends/bitcoind/blockchain_processor.py b/backends/bitcoind/blockchain_processor.py
index d4c56f4..3b65b27 100644
--- a/backends/bitcoind/blockchain_processor.py
+++ b/backends/bitcoind/blockchain_processor.py
@@ -37,7 +37,7 @@ class BlockchainProcessor(Processor):
 
         self.mempool_addresses = {}
         self.mempool_hist = {}
-        self.mempool_hashes = []
+        self.mempool_hashes = set([])
         self.mempool_lock = threading.Lock()
 
         self.address_queue = Queue()
@@ -686,14 +686,14 @@ class BlockchainProcessor(Processor):
         error = None
 
         if method == 'blockchain.numblocks.subscribe':
-            if session not in self.watch_headers:
-                with self.watch_lock:
+            with self.watch_lock:
+                if session not in self.watch_blocks:
                     self.watch_blocks.append(session)
             result = self.height
 
         elif method == 'blockchain.headers.subscribe':
-            if session not in self.watch_headers:
-                with self.watch_lock:
+            with self.watch_lock:
+                if session not in self.watch_headers:
                     self.watch_headers.append(session)
             result = self.header
 
@@ -863,10 +863,11 @@ class BlockchainProcessor(Processor):
 
         self.header = self.block2header(self.bitcoind('getblock', [self.last_hash]))
 
+
     def memorypool_update(self):
-        mempool_hashes = self.bitcoind('getrawmempool')
+        mempool_hashes = set(self.bitcoind('getrawmempool'))
+        touched_addresses = set([])
 
-        touched_addresses = []
         for tx_hash in mempool_hashes:
             if tx_hash in self.mempool_hashes:
                 continue
@@ -881,16 +882,16 @@ class BlockchainProcessor(Processor):
                 addr = x.get('address')
                 if addr and addr not in mpa:
                     mpa.append(addr)
-                    touched_addresses.append(addr)
+                    touched_addresses.add(addr)
 
             for x in tx.get('outputs'):
                 addr = x.get('address')
                 if addr and addr not in mpa:
                     mpa.append(addr)
-                    touched_addresses.append(addr)
+                    touched_addresses.add(addr)
 
             self.mempool_addresses[tx_hash] = mpa
-            self.mempool_hashes.append(tx_hash)
+            self.mempool_hashes.add(tx_hash)
 
         # remove older entries from mempool_hashes
         self.mempool_hashes = mempool_hashes
@@ -900,7 +901,7 @@ class BlockchainProcessor(Processor):
             if tx_hash not in self.mempool_hashes:
                 self.mempool_addresses.pop(tx_hash)
                 for addr in addresses:
-                    touched_addresses.append(addr)
+                    touched_addresses.add(addr)
 
         # rebuild mempool histories
         new_mempool_hist = {}
@@ -925,9 +926,12 @@ class BlockchainProcessor(Processor):
                 print_log("cache: invalidating", address)
                 self.history_cache.pop(address)
 
-        if address in self.watched_addresses:
+        with self.watch_lock:
+            sessions = self.watched_addresses.get(address)
+
+        if sessions:
             # TODO: update cache here. if new value equals cached value, do not send notification
-            self.address_queue.put(address)
+            self.address_queue.put((address,sessions))
 
     def main_iteration(self):
         if self.shared.stopped():
@@ -962,12 +966,12 @@ class BlockchainProcessor(Processor):
 
         while True:
             try:
-                addr = self.address_queue.get(False)
+                addr, sessions = self.address_queue.get(False)
             except:
                 break
 
             status = self.get_status(addr)
-            for session in self.watched_addresses[addr]:
+            for session in sessions:
                 self.push_response(session, {
                         'id': None,
                         'method': 'blockchain.address.subscribe',