Upgrade chaind to handle exception in filesystem shep list

.gitignore

@@ -6,3 +6,4 @@ build/
 *.egg-info
 *.sqlite
 .coverage
+.chaind

chaind/eth/runnable/queuer.py

@@ -28,11 +28,12 @@ from chainqueue.cache import CacheTokenTx
 from chainlib.encode import TxHexNormalizer
 from chainlib.chain import ChainSpec
 from chaind.adapters.fs import ChaindFsAdapter
+from chaind.dispatch import DispatchProcessor
 
 # local imports
-from chaind.eth.dispatch import EthDispatcher
 from chaind.eth.cache import EthCacheTx
 from chaind.eth.settings import ChaindEthSettings
+from chaind.eth.dispatch import EthDispatcher
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -66,56 +67,34 @@ settings.process(config)
 
 logg.debug('settings:\n{}'.format(settings))
 
-
-def process_outgoing(chain_spec, adapter, rpc, limit=50):
-    adapter = None
-    process_err = None
-    for i in range(2):
-        try:
-            adapter = ChaindFsAdapter(
-                settings.get('CHAIN_SPEC'),
-                settings.dir_for('queue'),
-                EthCacheTx,
-                dispatcher,
-                )
-        except BackendIntegrityError as e:
-            process_err = e
-            continue
-
-    if adapter == None:
-        raise BackendIntegrityError(process_err)
-    
-    upcoming = adapter.upcoming(limit=limit)
-    logg.info('processor has {} candidates for {}, processing with limit {} adapter {} rpc {}'.format(len(upcoming), chain_spec, limit, adapter, rpc))
-    i = 0
-    for tx_hash in upcoming:
-        if adapter.dispatch(tx_hash):
-            i += 1
-    return i
-
-chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
-
 rpc = chainlib.eth.cli.Rpc()
 conn = rpc.connect_by_config(config)
 
 tx_normalizer = TxHexNormalizer().tx_hash
-token_cache_store = CacheTokenTx(chain_spec, normalizer=tx_normalizer)
-dispatcher = EthDispatcher(conn)
-queue_adapter = ChaindFsAdapter(
-        settings.get('CHAIN_SPEC'),
-        settings.dir_for('queue'),
-        EthCacheTx,
-        dispatcher,
-        )
+token_cache_store = CacheTokenTx(settings.get('CHAIN_SPEC'), normalizer=tx_normalizer)
+
+dispatcher = EthDispatcher(conn)
+processor = DispatchProcessor(settings.get('CHAIN_SPEC'), settings.dir_for('queue'), dispatcher)
+ctrl = SessionController(settings, processor.process)
 
-ctrl = SessionController(settings, queue_adapter, process_outgoing)
 signal.signal(signal.SIGINT, ctrl.shutdown)
 signal.signal(signal.SIGTERM, ctrl.shutdown)
 
 logg.info('session id is ' + settings.get('SESSION_ID'))
 logg.info('session socket path is ' + settings.get('SESSION_SOCKET_PATH'))
 
+
 def main():
+    global dispatcher, settings
+
+    queue_adapter = ChaindFsAdapter(
+        settings.get('CHAIN_SPEC'),
+        settings.dir_for('queue'),
+        EthCacheTx,
+        dispatcher,
+        store_sync=False,
+        )
+
     while True:
         v = None
         client_socket = None
@@ -132,6 +111,7 @@ def main():
 
         if v == None:
             ctrl.process(conn)
+            #queue_adapter = create_adapter(settings, dispatcher)
             continue
 
         result_data = None

chaind/eth/runnable/send.py

@@ -118,7 +118,7 @@ def main():
         m = GasTokenResolver
     token_resolver = m(chain_spec, rpc.get_sender_address(), rpc.get_signer(), rpc.get_gas_oracle(), rpc.get_nonce_oracle())
     
-    processor = Processor(token_resolver, config.get('_SOURCE'))
+    processor = Processor(token_resolver, config.get('_SOURCE'), use_checksum=not config.get('_UNSAFE'))
     processor.add_processor(CSVProcessor())
 
     sends = None

chaind/eth/token/process.py

@@ -3,7 +3,10 @@ import logging
 
 # external imports
 from chaind.error import TxSourceError
-from chainlib.eth.address import is_checksum_address
+from chainlib.eth.address import (
+        is_checksum_address,
+        to_checksum_address,
+        )
 from chainlib.eth.tx import unpack
 from chainlib.eth.gas import Gas
 from hexathon import (
@@ -17,10 +20,11 @@ logg = logging.getLogger(__name__)
 
 class Processor:
 
-    def __init__(self, resolver, source):
+    def __init__(self, resolver, source, use_checksum=True):
         self.resolver = resolver
         self.source = source
         self.processor = []
+        self.safe = use_checksum
         self.conn = None
         
 
@@ -50,9 +54,14 @@ class Processor:
         txs = []
         for i, r in enumerate(self.content):
             logg.debug('processing {}'.format(r))
-            if not is_checksum_address(r[0]):
-                raise ValueError('invalid checksum address {} in record {}'.format(r[0], i))
-            self.content[i][0] = add_0x(r[0])
+            address = r[0]
+            if self.safe:
+                if not is_checksum_address(address):
+                    raise ValueError('invalid checksum address {} in record {}'.format(address, i))
+            else:
+                address = to_checksum_address(address)
+
+            self.content[i][0] = add_0x(address)
             try:
                 self.content[i][1] = int(r[1])
             except ValueError:

requirements.txt

@@ -1,6 +1,5 @@
-chaind~=0.2.0
-hexathon~=0.1.5
-chainlib-eth~=0.1.1
+chaind~=0.2.11
+hexathon~=0.1.6
+chainlib-eth~=0.1.2
 pyxdg~=0.27
-shep~=0.2.3
-funga-eth~=0.6.0
+funga-eth~=0.6.1

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chaind-eth
-version = 0.2.0
+version = 0.2.10
 description = Queue server for ethereum
 author = Louis Holbrook
 author_email = dev@holbrook.no

tests/test_tx.py

@@ -11,7 +11,7 @@ from chainlib.chain import ChainSpec
 from chainqueue.cache import CacheTokenTx
 from chainlib.error import RPCException
 from chainlib.status import Status as TxStatus
-from chaind.unittest.common import TestChaindFsBase
+from chaind.unittest.fs import TestChaindFsBase
 from chaind.driver import QueueDriver
 from chaind.filter import StateFilter
 from chainlib.eth.gas import Gas