From abc619d7b24493d76e4f89b78a18c88bbd07b1ac Mon Sep 17 00:00:00 2001
From: lash <dev@holbrook.no>
Date: Wed, 4 May 2022 05:41:00 +0000
Subject: [PATCH] Factor out outgoing processor

---
 .gitignore                    |  1 +
 chaind/eth/runnable/queuer.py | 46 ++++++++++++-----------------------
 requirements.txt              |  4 +--
 setup.cfg                     |  2 +-
 4 files changed, 19 insertions(+), 34 deletions(-)

diff --git a/.gitignore b/.gitignore
index fe8e060..4c6f42a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,3 +6,4 @@ build/
 *.egg-info
 *.sqlite
 .coverage
+.chaind
diff --git a/chaind/eth/runnable/queuer.py b/chaind/eth/runnable/queuer.py
index d464990..91abfa8 100644
--- a/chaind/eth/runnable/queuer.py
+++ b/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,41 +67,16 @@ settings.process(config)
 
 logg.debug('settings:\n{}'.format(settings))
 
-
-def process_outgoing(chain_spec, adapter, rpc, limit=50):
-    adapter = None
-    process_err = None
-    adapter = ChaindFsAdapter(
-        settings.get('CHAIN_SPEC'),
-        settings.dir_for('queue'),
-        EthCacheTx,
-        dispatcher,
-        )
-    
-    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)
 
@@ -126,6 +102,14 @@ def main():
             ctrl.process(conn)
             continue
 
+        queue_adapter = ChaindFsAdapter(
+            settings.get('CHAIN_SPEC'),
+            settings.dir_for('queue'),
+            EthCacheTx,
+            dispatcher,
+            store_sync=False,
+            )
+
         result_data = None
         r = 0 # no error
         try:
diff --git a/requirements.txt b/requirements.txt
index 2bfdc5c..817c6ee 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,6 +1,6 @@
-chaind~=0.2.2
+chaind~=0.2.4
 hexathon~=0.1.5
 chainlib-eth~=0.1.1
 pyxdg~=0.27
-shep~=0.2.3
+shep~=0.2.5
 funga-eth~=0.6.0
diff --git a/setup.cfg b/setup.cfg
index 96f8128..f384890 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,6 +1,6 @@
 [metadata]
 name = chaind-eth
-version = 0.2.1
+version = 0.2.2
 description = Queue server for ethereum
 author = Louis Holbrook
 author_email = dev@holbrook.no