chaintool/chainsyncer
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Stub for dynamic method/domain (task/queue) callback

Browse Source
This commit is contained in:
nolash 2021-02-03 23:32:19 +01:00
parent 991e909a2c
commit 8c67758b10
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
2 changed files with 22 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
cic_syncer/driver.py
View File

@ -10,11 +10,12 @@ class Syncer:
running_global = True running_global = True
def __init__(self, backend): def __init__(self, backend, handler):
self.cursor = None self.cursor = None
self.running = True self.running = True
self.backend = backend self.backend = backend
self.filter = [] self.filter = []
self.handler = handler
def chain(self): def chain(self):
@ -29,8 +30,8 @@ class Syncer:
class MinedSyncer(Syncer): class MinedSyncer(Syncer):
def __init__(self, backend): def __init__(self, backend, handler):
super(MinedSyncer, self).__init__(backend) super(MinedSyncer, self).__init__(backend, handler)
def loop(self, interval, getter): def loop(self, interval, getter):
@ -43,8 +44,8 @@ class MinedSyncer(Syncer):
class HeadSyncer(MinedSyncer): class HeadSyncer(MinedSyncer):
def __init__(self, backend): def __init__(self, backend, handler):
super(HeadSyncer, self).__init__(backend) super(HeadSyncer, self).__init__(backend, handler)
def process(self, getter, block): def process(self, getter, block):
@ -54,6 +55,7 @@ class HeadSyncer(MinedSyncer):
tx = None tx = None
while True: while True:
try: try:
self.filter[0].handle(getter, block, None)
tx = block.tx(i) tx = block.tx(i)
logg.debug('tx {}'.format(tx)) logg.debug('tx {}'.format(tx))
self.backend.set(block.number(), i) self.backend.set(block.number(), i)

18
cic_syncer/runnable/tracker.py
View File

@ -21,6 +21,18 @@ logg = logging.getLogger()
config_dir = '/usr/local/etc/cic-syncer' config_dir = '/usr/local/etc/cic-syncer'
class Handler:
def __init__(self, method, domain):
self.method = method
self.domain = domain
def handle(self, getter, tx, chain):
logg.debug('noop tx {} chain {} method {} domain {}'.format(tx, chain, self.method, self.domain))
handler = getattr(Handler, 'handle')
argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks') argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks')
argparser.add_argument('-p', '--provider', dest='p', type=str, help='chain rpc provider address') argparser.add_argument('-p', '--provider', dest='p', type=str, help='chain rpc provider address')
argparser.add_argument('-c', type=str, default=config_dir, help='config root to use') argparser.add_argument('-c', type=str, default=config_dir, help='config root to use')
@ -58,7 +70,6 @@ queue = args.q
dsn = dsn_from_config(config) dsn = dsn_from_config(config)
SessionBase.connect(dsn) SessionBase.connect(dsn)
# TODO: There is too much code in this file, split it up
transfer_callbacks = [] transfer_callbacks = []
for cb in config.get('TASKS_SYNCER_CALLBACKS', '').split(','): for cb in config.get('TASKS_SYNCER_CALLBACKS', '').split(','):
@ -100,7 +111,7 @@ def main():
block_offset = c.block_number() block_offset = c.block_number()
syncer_backend = SyncerBackend.live(chain, block_offset+1) syncer_backend = SyncerBackend.live(chain, block_offset+1)
syncer = HeadSyncer(syncer_backend) syncer = HeadSyncer(syncer_backend, handler)
for cb in config.get('TASKS_SYNCER_CALLBACKS', '').split(','): for cb in config.get('TASKS_SYNCER_CALLBACKS', '').split(','):
task_split = cb.split(':') task_split = cb.split(':')
@ -108,7 +119,8 @@ def main():
if len(task_split) > 1: if len(task_split) > 1:
task_queue = task_split[0] task_queue = task_split[0]
task_pair = (task_split[1], task_queue) task_pair = (task_split[1], task_queue)
syncer.filter.append(task_pair) h = Handler(task_pair[0], task_pair[1])
syncer.filter.append(h)
try: try:
logg.debug('block offset {} {}'.format(block_offset, c)) logg.debug('block offset {} {}'.format(block_offset, c))