WIP implement chainqueue in daemons

This commit is contained in:
nolash 2021-04-03 00:16:31 +02:00
parent ffe7416040
commit 08fba819e4
Signed by untrusted user who does not match committer: lash
GPG Key ID: 21D2E7BB88C2A746
3 changed files with 17 additions and 18 deletions

View File

@ -17,26 +17,25 @@ from chainlib.eth.tx import unpack
from chainlib.connection import RPCConnection from chainlib.connection import RPCConnection
from chainsyncer.error import SyncDone from chainsyncer.error import SyncDone
from hexathon import strip_0x from hexathon import strip_0x
from chainqueue.db.enum import (
StatusEnum,
StatusBits,
)
from chainqueue.error import NotLocalTxError
# local imports # local imports
import cic_eth import cic_eth
from cic_eth.db import SessionBase from cic_eth.db import SessionBase
from cic_eth.db.enum import StatusEnum
from cic_eth.db.enum import StatusBits
from cic_eth.db.enum import LockEnum from cic_eth.db.enum import LockEnum
from cic_eth.db import dsn_from_config from cic_eth.db import dsn_from_config
from cic_eth.queue.tx import ( from cic_eth.queue.query import get_upcoming_tx
get_upcoming_tx, from cic_eth.queue.state import set_reserved
set_dequeue,
)
from cic_eth.admin.ctrl import lock_send from cic_eth.admin.ctrl import lock_send
from cic_eth.eth.tx import send as task_tx_send from cic_eth.eth.tx import send as task_tx_send
from cic_eth.error import ( from cic_eth.error import (
PermanentTxError, PermanentTxError,
TemporaryTxError, TemporaryTxError,
NotLocalTxError,
) )
#from cic_eth.eth.util import unpack_signed_raw_tx_hex
logging.basicConfig(level=logging.WARNING) logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger() logg = logging.getLogger()
@ -140,7 +139,7 @@ class DispatchSyncer:
while run: while run:
txs = {} txs = {}
typ = StatusBits.QUEUED typ = StatusBits.QUEUED
utxs = get_upcoming_tx(typ, chain_id=self.chain_id) utxs = get_upcoming_tx(self.chain_spec, typ)
for k in utxs.keys(): for k in utxs.keys():
txs[k] = utxs[k] txs[k] = utxs[k]
self.process(w3, txs) self.process(w3, txs)

View File

@ -3,13 +3,13 @@ import logging
# external imports # external imports
from hexathon import add_0x from hexathon import add_0x
from chainqueue.db.enum import StatusBits
from chainqueue.db.models.tx import TxCache
from chainqueue.db.models.otx import Otx
from chainqueue.query import get_paused_tx_cache as get_paused_tx
# local imports # local imports
from cic_eth.db.enum import StatusBits
from cic_eth.db.models.base import SessionBase from cic_eth.db.models.base import SessionBase
from cic_eth.db.models.tx import TxCache
from cic_eth.db.models.otx import Otx
from cic_eth.queue.tx import get_paused_txs
from cic_eth.eth.gas import create_check_gas_task from cic_eth.eth.gas import create_check_gas_task
from .base import SyncFilter from .base import SyncFilter
@ -38,7 +38,7 @@ class GasFilter(SyncFilter):
SessionBase.release_session(session) SessionBase.release_session(session)
return return
txs = get_paused_txs(StatusBits.GAS_ISSUES, r[0], self.chain_spec.chain_id(), session=session) txs = get_paused_tx(self.chain_spec, StatusBits.GAS_ISSUES, r[0], self.chain_spec.chain_id(), session=session)
SessionBase.release_session(session) SessionBase.release_session(session)

View File

@ -1,16 +1,16 @@
# standard imports # standard imports
import logging import logging
# third-party imports # external imports
import celery import celery
from hexathon import ( from hexathon import (
add_0x, add_0x,
) )
from chainsyncer.db.models.base import SessionBase
from chainqueue.db.models.otx import Otx
from chainlib.status import Status
# local imports # local imports
from cic_eth.db.models.otx import Otx
from chainsyncer.db.models.base import SessionBase
from chainlib.status import Status
from .base import SyncFilter from .base import SyncFilter
logg = logging.getLogger().getChild(__name__) logg = logging.getLogger().getChild(__name__)