Merge remote-tracking branch 'origin/develop' into 291_proxy_refactoring

proxy/indexer/indexer.py

@@ -5,12 +5,8 @@
 import os
 import rlp
 import time
-
+import logging
 from multiprocessing.dummy import Pool as ThreadPool
-from solana.rpc.api import Client
-from typing import Dict, Union
-
-from ..environment import SOLANA_URL, EVM_LOADER_ID
 
 
 try:
@@ -51,9 +47,13 @@ def __init__(self, eth_trx, eth_signature, from_address, got_result, signatures,
         self.slot = slot
 
 
-class Indexer:
-    def __init__(self):
-        self.client = Client(SOLANA_URL)
+class Indexer(IndexerBase):
+    def __init__(self,
+                 solana_url,
+                 evm_loader_id,
+                 log_level = 'INFO'):
+        IndexerBase.__init__(self, solana_url, evm_loader_id, log_level)
+
         self.canceller = Canceller()
         self.logs_db = LogDB()
         self.blocks_by_hash = SQLDict(tablename="solana_blocks_by_hash")
@@ -66,96 +66,15 @@ def __init__(self):
         self.blocked_storages = {}
 
 
-    def gather_unknown_transactions(self):
-        poll_txs = set()
-        ordered_txs = []
-
-        minimal_tx = None
-        continue_flag = True
-        current_slot = self.client.get_slot(commitment="confirmed")["result"]
-        maximum_slot = self.last_slot
-        minimal_slot = current_slot
-
-        percent = 0
-
-        counter = 0
-        while (continue_flag):
-            opts: Dict[str, Union[int, str]] = {}
-            if minimal_tx:
-                opts["before"] = minimal_tx
-            opts["commitment"] = "confirmed"
-            result = self.client._provider.make_request("getSignaturesForAddress", EVM_LOADER_ID, opts)
-            logger.debug("{:>3} get_signatures_for_address {}".format(counter, len(result["result"])))
-            counter += 1
-
-            if len(result["result"]) == 0:
-                logger.debug("len(result['result']) == 0")
-                break
-
-            for tx in result["result"]:
-                solana_signature = tx["signature"]
-                slot = tx["slot"]
-
-                if solana_signature in HISTORY_START:
-                    logger.debug(solana_signature)
-                    continue_flag = False
-                    break
-
-                ordered_txs.append(solana_signature)
-
-                if solana_signature not in self.transaction_receipts:
-                    poll_txs.add(solana_signature)
-
-                if slot < minimal_slot:
-                    minimal_slot = slot
-                    minimal_tx = solana_signature
-
-                if slot > maximum_slot:
-                    maximum_slot = slot
-
-                if slot < self.last_slot:
-                    continue_flag = False
-                    break
-
-        logger.debug("start getting receipts")
-        pool = ThreadPool(PARALLEL_REQUESTS)
-        pool.map(self.get_tx_receipts, poll_txs)
-
-        if len(self.transaction_order):
-            index = 0
-            try:
-                index = ordered_txs.index(self.transaction_order[0])
-            except ValueError:
-                self.transaction_order = ordered_txs + self.transaction_order
-            else:
-                self.transaction_order = ordered_txs[:index] + self.transaction_order
-        else:
-            self.transaction_order = ordered_txs
-
-        self.last_slot = maximum_slot
-        self.current_slot = current_slot
-
-        self.counter_ = 0
-
-
-    def get_tx_receipts(self, solana_signature):
-        # trx = None
-        retry = True
-
-        while retry:
-            try:
-                trx = self.client.get_confirmed_transaction(solana_signature)['result']
-                self.transaction_receipts[solana_signature] = trx
-                retry = False
-            except Exception as err:
-                logger.debug(err)
-                time.sleep(1)
-
-        self.counter_ += 1
-        if self.counter_ % 100 == 0:
-            logger.debug(self.counter_)
-
-        # return (solana_signature, trx)
+    def process_functions(self):
+        IndexerBase.process_functions(self)
+        logger.debug("Process receipts")
+        self.process_receipts()
+        logger.debug("Start getting blocks")
+        self.gather_blocks()
+        logger.debug("Unlock accounts")
+        self.canceller.unlock_accounts(self.blocked_storages)
+        self.blocked_storages = {}
 
 
     def process_receipts(self):
@@ -183,7 +102,7 @@ def process_receipts(self):
                 if trx['transaction']['message']['instructions'] is not None:
                     for instruction in trx['transaction']['message']['instructions']:
 
-                        if trx["transaction"]["message"]["accountKeys"][instruction["programIdIndex"]] != EVM_LOADER_ID:
+                        if trx["transaction"]["message"]["accountKeys"][instruction["programIdIndex"]] != self.evm_loader_id:
                             continue
 
                         if check_error(trx):