Compare commits
No commits in common. "master" and "dev-0.3.7" have entirely different histories.
|
@ -141,4 +141,3 @@ class SyncDriver:
|
||||||
|
|
||||||
def get(self, conn):
|
def get(self, conn):
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
|
|
|
@ -43,17 +43,13 @@ class FilterState:
|
||||||
self.state_store.add('RESET')
|
self.state_store.add('RESET')
|
||||||
|
|
||||||
self.state = self.state_store.state
|
self.state = self.state_store.state
|
||||||
self.elements = self.state_store.elements
|
|
||||||
self.put = self.state_store.put
|
self.put = self.state_store.put
|
||||||
self.mask = self.state_store.mask
|
|
||||||
self.name = self.state_store.name
|
|
||||||
self.set = self.state_store.set
|
self.set = self.state_store.set
|
||||||
self.next = self.state_store.next
|
self.next = self.state_store.next
|
||||||
self.move = self.state_store.move
|
self.move = self.state_store.move
|
||||||
self.unset = self.state_store.unset
|
self.unset = self.state_store.unset
|
||||||
self.peek = self.state_store.peek
|
self.peek = self.state_store.peek
|
||||||
self.from_name = self.state_store.from_name
|
self.from_name = self.state_store.from_name
|
||||||
self.list = self.state_store.list
|
|
||||||
self.state_store.sync()
|
self.state_store.sync()
|
||||||
self.all = self.state_store.all
|
self.all = self.state_store.all
|
||||||
self.started = False
|
self.started = False
|
||||||
|
@ -97,7 +93,6 @@ class FilterState:
|
||||||
if self.scan != None:
|
if self.scan != None:
|
||||||
ks = self.scan()
|
ks = self.scan()
|
||||||
for v in ks: #os.listdir(self.scan_path):
|
for v in ks: #os.listdir(self.scan_path):
|
||||||
logg.debug('ks {}'.format(v))
|
|
||||||
k = None
|
k = None
|
||||||
try:
|
try:
|
||||||
k = self.state_store.from_elements(v)
|
k = self.state_store.from_elements(v)
|
||||||
|
|
|
@ -0,0 +1,66 @@
|
||||||
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
|
||||||
|
# standard imports
|
||||||
|
import os
|
||||||
|
import logging
|
||||||
|
import sys
|
||||||
|
import importlib
|
||||||
|
|
||||||
|
# external imports
|
||||||
|
import chainlib.cli
|
||||||
|
|
||||||
|
# local imports
|
||||||
|
import chainsyncer.cli
|
||||||
|
from chainsyncer.settings import ChainsyncerSettings
|
||||||
|
|
||||||
|
logging.basicConfig(level=logging.WARNING)
|
||||||
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
arg_flags = chainlib.cli.argflag_std_base | chainlib.cli.Flag.CHAIN_SPEC
|
||||||
|
argparser = chainlib.cli.ArgumentParser(arg_flags)
|
||||||
|
argparser.add_argument('--state-dir', type=str, dest='state_dir', help='State directory')
|
||||||
|
argparser.add_argument('--session-id', type=str, dest='session_id', help='Session id for state')
|
||||||
|
argparser.add_argument('--action', type=str, choices=['repeat', 'continue'], help='Action to take on lock. Repeat means re-run the locked filter. Continue means resume execution for next filter.')
|
||||||
|
argparser.add_positional('tx', type=str, help='Transaction hash to unlock')
|
||||||
|
|
||||||
|
sync_flags = chainsyncer.cli.SyncFlag.RANGE | chainsyncer.cli.SyncFlag.HEAD
|
||||||
|
chainsyncer.cli.process_flags(argparser, sync_flags)
|
||||||
|
|
||||||
|
args = argparser.parse_args()
|
||||||
|
|
||||||
|
base_config_dir = chainsyncer.cli.config_dir,
|
||||||
|
config = chainlib.cli.Config.from_args(args, arg_flags, base_config_dir=base_config_dir)
|
||||||
|
config = chainsyncer.cli.process_config(config, args, sync_flags)
|
||||||
|
config.add(args.state_dir, '_STATE_DIR', False)
|
||||||
|
config.add(args.session_id, '_SESSION_ID', False)
|
||||||
|
logg.debug('config loaded:\n{}'.format(config))
|
||||||
|
|
||||||
|
settings = ChainsyncerSettings()
|
||||||
|
settings.process_sync_backend(config)
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
state_dir = None
|
||||||
|
if settings.get('SYNCER_BACKEND') == 'mem':
|
||||||
|
raise ValueError('cannot unlock volatile state store')
|
||||||
|
|
||||||
|
if settings.get('SYNCER_BACKEND') == 'fs':
|
||||||
|
syncer_store_module = importlib.import_module('chainsyncer.store.fs')
|
||||||
|
syncer_store_class = getattr(syncer_store_module, 'SyncFsStore')
|
||||||
|
elif settings.get('SYNCER_BACKEND') == 'rocksdb':
|
||||||
|
syncer_store_module = importlib.import_module('chainsyncer.store.rocksdb')
|
||||||
|
syncer_store_class = getattr(syncer_store_module, 'SyncRocksDbStore')
|
||||||
|
else:
|
||||||
|
syncer_store_module = importlib.import_module(settings.get('SYNCER_BACKEND'))
|
||||||
|
syncer_store_class = getattr(syncer_store_module, 'SyncStore')
|
||||||
|
|
||||||
|
state_dir = os.path.join(config.get('_STATE_DIR'), settings.get('SYNCER_BACKEND'))
|
||||||
|
sync_path = os.path.join(config.get('_SESSION_ID'), 'sync', 'filter')
|
||||||
|
sync_store = syncer_store_class(state_dir, session_id=sync_path)
|
||||||
|
|
||||||
|
logg.info('session is {}'.format(sync_store.session_id))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
main()
|
|
@ -1,146 +0,0 @@
|
||||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
||||||
|
|
||||||
# standard imports
|
|
||||||
import os
|
|
||||||
import logging
|
|
||||||
import sys
|
|
||||||
import importlib
|
|
||||||
|
|
||||||
# external imports
|
|
||||||
import chainlib.cli
|
|
||||||
from shep.persist import PersistedState
|
|
||||||
|
|
||||||
# local imports
|
|
||||||
import chainsyncer.cli
|
|
||||||
from chainsyncer.settings import ChainsyncerSettings
|
|
||||||
from chainsyncer.store import SyncStore
|
|
||||||
from chainsyncer.filter import (
|
|
||||||
FilterState,
|
|
||||||
SyncFilter,
|
|
||||||
)
|
|
||||||
|
|
||||||
logging.basicConfig(level=logging.WARNING)
|
|
||||||
logg = logging.getLogger()
|
|
||||||
|
|
||||||
valid_fwd = [
|
|
||||||
'fwd',
|
|
||||||
'forward',
|
|
||||||
'next',
|
|
||||||
'continue',
|
|
||||||
]
|
|
||||||
|
|
||||||
valid_rwd = [
|
|
||||||
'rwd',
|
|
||||||
'rewind',
|
|
||||||
'current',
|
|
||||||
'back',
|
|
||||||
'repeat',
|
|
||||||
'replay',
|
|
||||||
]
|
|
||||||
|
|
||||||
action_is_forward = False
|
|
||||||
|
|
||||||
arg_flags = chainlib.cli.argflag_std_base | chainlib.cli.Flag.CHAIN_SPEC
|
|
||||||
argparser = chainlib.cli.ArgumentParser(arg_flags)
|
|
||||||
argparser.add_argument('--state-dir', type=str, dest='state_dir', help='State directory')
|
|
||||||
argparser.add_positional('action', type=str, help='Action to take on lock. Repeat means re-run the locked filter. Continue means resume execution for next filter.')
|
|
||||||
|
|
||||||
sync_flags = chainsyncer.cli.SyncFlag.RANGE | chainsyncer.cli.SyncFlag.HEAD
|
|
||||||
chainsyncer.cli.process_flags(argparser, sync_flags)
|
|
||||||
|
|
||||||
args = argparser.parse_args()
|
|
||||||
|
|
||||||
if args.action in valid_fwd:
|
|
||||||
action_is_forward = True
|
|
||||||
elif args.action not in valid_rwd:
|
|
||||||
sys.stderr.write('action argument must be one of {} or {}\n'.format(valid_rwd, valid_fwd))
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
|
|
||||||
base_config_dir = chainsyncer.cli.config_dir,
|
|
||||||
config = chainlib.cli.Config.from_args(args, arg_flags, base_config_dir=base_config_dir)
|
|
||||||
config = chainsyncer.cli.process_config(config, args, sync_flags)
|
|
||||||
config.add(args.state_dir, '_STATE_DIR', False)
|
|
||||||
logg.debug('config loaded:\n{}'.format(config))
|
|
||||||
|
|
||||||
settings = ChainsyncerSettings()
|
|
||||||
settings.process_sync_backend(config)
|
|
||||||
logg.debug('settings:\n{}'.format(str(settings)))
|
|
||||||
|
|
||||||
|
|
||||||
class FilterInNameOnly(SyncFilter):
|
|
||||||
|
|
||||||
def __init__(self, k):
|
|
||||||
self.k = k
|
|
||||||
|
|
||||||
|
|
||||||
def common_name(self):
|
|
||||||
return self.k
|
|
||||||
|
|
||||||
|
|
||||||
def main():
|
|
||||||
if settings.get('SYNCER_BACKEND') == 'mem':
|
|
||||||
raise ValueError('cannot unlock volatile state store')
|
|
||||||
|
|
||||||
state_dir = config.get('_STATE_DIR')
|
|
||||||
|
|
||||||
if config.get('SYNCER_BACKEND') == 'fs':
|
|
||||||
syncer_store_module = importlib.import_module('chainsyncer.store.fs')
|
|
||||||
syncer_store_class = getattr(syncer_store_module, 'SyncFsStore')
|
|
||||||
elif config.get('SYNCER_BACKEND') == 'rocksdb':
|
|
||||||
syncer_store_module = importlib.import_module('chainsyncer.store.rocksdb')
|
|
||||||
syncer_store_class = getattr(syncer_store_module, 'SyncRocksDbStore')
|
|
||||||
else:
|
|
||||||
syncer_store_module = importlib.import_module(config.get('SYNCER_BACKEND'))
|
|
||||||
syncer_store_class = getattr(syncer_store_module, 'SyncStore')
|
|
||||||
|
|
||||||
logg.info('using engine {} module {}.{}'.format(config.get('SYNCER_BACKEND'), syncer_store_module.__file__, syncer_store_class.__name__))
|
|
||||||
|
|
||||||
store = syncer_store_class(state_dir)
|
|
||||||
|
|
||||||
filter_list = store.load_filter_list()
|
|
||||||
for i, k in enumerate(filter_list):
|
|
||||||
fltr = FilterInNameOnly(k)
|
|
||||||
store.register(fltr)
|
|
||||||
filter_list[i] = k.upper()
|
|
||||||
|
|
||||||
store.connect()
|
|
||||||
store.start(ignore_lock=True)
|
|
||||||
|
|
||||||
lock_state = store.filter_state.from_name('LOCK')
|
|
||||||
locked_item = store.filter_state.list(lock_state)
|
|
||||||
if len(locked_item) == 0:
|
|
||||||
sys.stderr.write('Sync filter in {} is not locked\n'.format(state_dir))
|
|
||||||
sys.exit(1)
|
|
||||||
elif len(locked_item) > 1:
|
|
||||||
sys.stderr.write('More than one locked item encountered in {}. That should never happen, so I do not know what to do next.\n'.format(state_dir))
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
locked_item_key = locked_item[0]
|
|
||||||
locked_item = store.get(int(locked_item_key))
|
|
||||||
locked_state = store.filter_state.state(locked_item_key) - lock_state
|
|
||||||
locked_state_name = store.filter_state.name(locked_state)
|
|
||||||
logg.info('found item "{}" in locked state {}'.format(locked_item, store.filter_state.name(locked_state)))
|
|
||||||
|
|
||||||
if action_is_forward:
|
|
||||||
k = locked_state_name
|
|
||||||
filter_index = None
|
|
||||||
filter_index = filter_list.index(k)
|
|
||||||
filter_pos = filter_index + 1
|
|
||||||
filter_count = len(filter_list)
|
|
||||||
logg.debug('Locked filter {} found at position {} of {}'.format(k, filter_pos, filter_count))
|
|
||||||
if filter_pos == filter_count:
|
|
||||||
logg.info('Locked filter {} is the last filter in the list. Executing filter reset'.format(k))
|
|
||||||
locked_item.reset(check_incomplete=False)
|
|
||||||
else:
|
|
||||||
locked_item.advance(ignore_lock=True)
|
|
||||||
store.filter_state.unset(locked_item_key, lock_state)
|
|
||||||
else:
|
|
||||||
filter_mask = 0xf
|
|
||||||
filter_state = store.filter_state.mask(locked_state, filter_mask)
|
|
||||||
logg.info('Chosen action is "{}": will continue execution at previous filter {}'.format(args.action, store.filter_state.name(filter_state)))
|
|
||||||
store.filter_state.unset(locked_item_key, lock_state)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
main()
|
|
|
@ -6,12 +6,16 @@ from hexathon import (
|
||||||
to_int as hex_to_int,
|
to_int as hex_to_int,
|
||||||
strip_0x,
|
strip_0x,
|
||||||
)
|
)
|
||||||
from chainlib.settings import ChainSettings
|
|
||||||
|
|
||||||
logg = logging.getLogger(__name__)
|
logg = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class ChainsyncerSettings(ChainSettings):
|
class ChainsyncerSettings:
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.o = {}
|
||||||
|
self.get = self.o.get
|
||||||
|
|
||||||
|
|
||||||
def process_sync_backend(self, config):
|
def process_sync_backend(self, config):
|
||||||
self.o['SYNCER_BACKEND'] = config.get('SYNCER_BACKEND')
|
self.o['SYNCER_BACKEND'] = config.get('SYNCER_BACKEND')
|
||||||
|
|
|
@ -35,7 +35,7 @@ def sync_state_deserialize(b):
|
||||||
# NOT thread safe
|
# NOT thread safe
|
||||||
class SyncItem:
|
class SyncItem:
|
||||||
|
|
||||||
def __init__(self, offset, target, sync_state, filter_state, started=False, ignore_lock=False):
|
def __init__(self, offset, target, sync_state, filter_state, started=False, ignore_invalid=False):
|
||||||
self.offset = offset
|
self.offset = offset
|
||||||
self.target = target
|
self.target = target
|
||||||
self.sync_state = sync_state
|
self.sync_state = sync_state
|
||||||
|
@ -47,8 +47,7 @@ class SyncItem:
|
||||||
|
|
||||||
(self.cursor, self.tx_cursor, self.target) = sync_state_deserialize(v)
|
(self.cursor, self.tx_cursor, self.target) = sync_state_deserialize(v)
|
||||||
|
|
||||||
filter_state = self.filter_state.state(self.state_key)
|
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') and not ignore_invalid:
|
||||||
if filter_state & self.filter_state.from_name('LOCK') > 0 and not ignore_lock:
|
|
||||||
raise LockError(self.state_key)
|
raise LockError(self.state_key)
|
||||||
|
|
||||||
self.count = len(self.filter_state.all(pure=True)) - 4
|
self.count = len(self.filter_state.all(pure=True)) - 4
|
||||||
|
@ -57,7 +56,7 @@ class SyncItem:
|
||||||
self.skip_filter = True
|
self.skip_filter = True
|
||||||
elif not started:
|
elif not started:
|
||||||
self.filter_state.move(self.state_key, self.filter_state.from_name('RESET'))
|
self.filter_state.move(self.state_key, self.filter_state.from_name('RESET'))
|
||||||
|
|
||||||
|
|
||||||
def __check_done(self):
|
def __check_done(self):
|
||||||
if self.filter_state.state(self.state_key) & self.filter_state.from_name('INTERRUPT') > 0:
|
if self.filter_state.state(self.state_key) & self.filter_state.from_name('INTERRUPT') > 0:
|
||||||
|
@ -66,22 +65,11 @@ class SyncItem:
|
||||||
raise FilterDone(self.state_key)
|
raise FilterDone(self.state_key)
|
||||||
|
|
||||||
|
|
||||||
def resume(self):
|
def reset(self):
|
||||||
filter_state = self.filter_state.state(self.state_key)
|
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
|
||||||
if filter_state > 0x0f:
|
raise LockError('reset attempt on {} when state locked'.format(self.state_key))
|
||||||
filter_state_part = self.filter_state.mask(filter_state, 0x0f)
|
if self.filter_state.state(self.state_key) & self.filter_state.from_name('DONE') == 0:
|
||||||
if len(self.filter_state.elements(filter_state)) == 1:
|
raise IncompleteFilterError('reset attempt on {} when incomplete'.format(self.state_key))
|
||||||
logg.info('resume execution on state {} ({})'.format(self.filter_state.name(filter_state_part), filter_state_part))
|
|
||||||
lock_state = self.filter_state.from_name('LOCK')
|
|
||||||
self.filter_state.set(lock_state)
|
|
||||||
|
|
||||||
|
|
||||||
def reset(self, check_incomplete=True):
|
|
||||||
if check_incomplete:
|
|
||||||
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
|
|
||||||
raise LockError('reset attempt on {} when state locked'.format(self.state_key))
|
|
||||||
if self.filter_state.state(self.state_key) & self.filter_state.from_name('DONE') == 0:
|
|
||||||
raise IncompleteFilterError('reset attempt on {} when incomplete'.format(self.state_key))
|
|
||||||
self.filter_state.move(self.state_key, self.filter_state.from_name('RESET'))
|
self.filter_state.move(self.state_key, self.filter_state.from_name('RESET'))
|
||||||
|
|
||||||
|
|
||||||
|
@ -114,16 +102,13 @@ class SyncItem:
|
||||||
v = self.filter_state.state(self.state_key)
|
v = self.filter_state.state(self.state_key)
|
||||||
|
|
||||||
|
|
||||||
def advance(self, ignore_lock=False):
|
def advance(self):
|
||||||
if self.skip_filter:
|
if self.skip_filter:
|
||||||
raise FilterDone()
|
raise FilterDone()
|
||||||
self.__check_done()
|
self.__check_done()
|
||||||
|
|
||||||
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
|
if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
|
||||||
if ignore_lock:
|
raise LockError('advance attempt on {} when state locked'.format(self.state_key))
|
||||||
self.filter_state.unset(self.state_key, self.filter_state.from_name('LOCK'))
|
|
||||||
else:
|
|
||||||
raise LockError('advance attempt on {} when state locked'.format(self.state_key))
|
|
||||||
done = False
|
done = False
|
||||||
try:
|
try:
|
||||||
self.filter_state.next(self.state_key)
|
self.filter_state.next(self.state_key)
|
||||||
|
@ -163,7 +148,7 @@ class SyncItem:
|
||||||
class SyncStore:
|
class SyncStore:
|
||||||
|
|
||||||
def __init__(self, path, session_id=None):
|
def __init__(self, path, session_id=None):
|
||||||
self.session_id = session_id
|
self.session_id = None
|
||||||
self.session_path = None
|
self.session_path = None
|
||||||
self.is_default = False
|
self.is_default = False
|
||||||
self.first = False
|
self.first = False
|
||||||
|
@ -172,7 +157,16 @@ class SyncStore:
|
||||||
self.item_keys = []
|
self.item_keys = []
|
||||||
self.started = False
|
self.started = False
|
||||||
self.thresholds = []
|
self.thresholds = []
|
||||||
self.session_path = path
|
self.default_path = os.path.join(path, 'default')
|
||||||
|
|
||||||
|
if session_id == None:
|
||||||
|
self.session_path = os.path.realpath(self.default_path)
|
||||||
|
self.is_default = True
|
||||||
|
else:
|
||||||
|
if session_id == 'default':
|
||||||
|
self.is_default = True
|
||||||
|
given_path = os.path.join(path, session_id)
|
||||||
|
self.session_path = os.path.realpath(given_path)
|
||||||
|
|
||||||
|
|
||||||
def setup_sync_state(self, factory=None, event_callback=None):
|
def setup_sync_state(self, factory=None, event_callback=None):
|
||||||
|
@ -207,21 +201,18 @@ class SyncStore:
|
||||||
self.filter_state.register(fltr)
|
self.filter_state.register(fltr)
|
||||||
|
|
||||||
|
|
||||||
def start(self, offset=0, target=-1, ignore_lock=False):
|
def start(self, offset=0, target=-1):
|
||||||
if self.started:
|
if self.started:
|
||||||
return
|
return
|
||||||
|
|
||||||
self.save_filter_list()
|
self.load(target)
|
||||||
|
|
||||||
self.load(target, ignore_lock=ignore_lock)
|
|
||||||
|
|
||||||
if self.first:
|
if self.first:
|
||||||
state_bytes = sync_state_serialize(offset, 0, target)
|
state_bytes = sync_state_serialize(offset, 0, target)
|
||||||
block_number_str = str(offset)
|
block_number_str = str(offset)
|
||||||
self.state.put(block_number_str, contents=state_bytes)
|
self.state.put(block_number_str, contents=state_bytes)
|
||||||
self.filter_state.put(block_number_str)
|
self.filter_state.put(block_number_str)
|
||||||
o = SyncItem(offset, target, self.state, self.filter_state, ignore_lock=ignore_lock)
|
o = SyncItem(offset, target, self.state, self.filter_state)
|
||||||
o.resume()
|
|
||||||
self.items[offset] = o
|
self.items[offset] = o
|
||||||
self.item_keys.append(offset)
|
self.item_keys.append(offset)
|
||||||
elif offset > 0:
|
elif offset > 0:
|
||||||
|
@ -246,7 +237,7 @@ class SyncStore:
|
||||||
self.state.put(str(item.cursor), contents=state_bytes)
|
self.state.put(str(item.cursor), contents=state_bytes)
|
||||||
|
|
||||||
|
|
||||||
def load(self, target, ignore_lock=False):
|
def load(self, target):
|
||||||
self.state.sync(self.state.NEW)
|
self.state.sync(self.state.NEW)
|
||||||
self.state.sync(self.state.SYNC)
|
self.state.sync(self.state.SYNC)
|
||||||
|
|
||||||
|
@ -270,8 +261,7 @@ class SyncStore:
|
||||||
item_target = target
|
item_target = target
|
||||||
if i < lim:
|
if i < lim:
|
||||||
item_target = thresholds[i+1]
|
item_target = thresholds[i+1]
|
||||||
o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True, ignore_lock=ignore_lock)
|
o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True)
|
||||||
o.resume()
|
|
||||||
self.items[block_number] = o
|
self.items[block_number] = o
|
||||||
self.item_keys.append(block_number)
|
self.item_keys.append(block_number)
|
||||||
logg.info('added existing {}'.format(o))
|
logg.info('added existing {}'.format(o))
|
||||||
|
@ -305,18 +295,3 @@ class SyncStore:
|
||||||
|
|
||||||
def disconnect(self):
|
def disconnect(self):
|
||||||
self.filter_state.disconnect()
|
self.filter_state.disconnect()
|
||||||
|
|
||||||
|
|
||||||
def save_filter_list(self):
|
|
||||||
raise NotImplementedError()
|
|
||||||
|
|
||||||
|
|
||||||
def load_filter_list(self):
|
|
||||||
raise NotImplementedError()
|
|
||||||
|
|
||||||
|
|
||||||
def peek_next_filter(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
def peek_current_filter(self):
|
|
||||||
pass
|
|
||||||
|
|
|
@ -74,25 +74,3 @@ class SyncFsStore(SyncStore):
|
||||||
f.write(str(v))
|
f.write(str(v))
|
||||||
f.close()
|
f.close()
|
||||||
self.target = v
|
self.target = v
|
||||||
|
|
||||||
|
|
||||||
def load_filter_list(self):
|
|
||||||
fltr = []
|
|
||||||
fp = os.path.join(self.session_path, 'filter_list')
|
|
||||||
f = open(fp, 'r')
|
|
||||||
while True:
|
|
||||||
v = f.readline()
|
|
||||||
if len(v) == 0:
|
|
||||||
break
|
|
||||||
v = v.rstrip()
|
|
||||||
fltr.append(v)
|
|
||||||
f.close()
|
|
||||||
return fltr
|
|
||||||
|
|
||||||
|
|
||||||
def save_filter_list(self):
|
|
||||||
fp = os.path.join(self.session_path, 'filter_list')
|
|
||||||
f = open(fp, 'w')
|
|
||||||
for fltr in self.filters:
|
|
||||||
f.write(fltr.common_name() + '\n')
|
|
||||||
f.close()
|
|
||||||
|
|
|
@ -14,7 +14,10 @@ logg = logging.getLogger(__name__)
|
||||||
class SyncMemStore(SyncStore):
|
class SyncMemStore(SyncStore):
|
||||||
|
|
||||||
def __init__(self, session_id=None, state_event_callback=None, filter_state_event_callback=None):
|
def __init__(self, session_id=None, state_event_callback=None, filter_state_event_callback=None):
|
||||||
super(SyncMemStore, self).__init__(None, session_id=session_id)
|
super(SyncMemStore, self).__init__('/dev/null', session_id=session_id)
|
||||||
|
|
||||||
|
self.session_id = os.path.basename(self.session_path)
|
||||||
|
logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))
|
||||||
|
|
||||||
factory = None
|
factory = None
|
||||||
self.setup_sync_state(factory, state_event_callback)
|
self.setup_sync_state(factory, state_event_callback)
|
||||||
|
@ -33,13 +36,5 @@ class SyncMemStore(SyncStore):
|
||||||
|
|
||||||
def stop(self, item):
|
def stop(self, item):
|
||||||
if item != None:
|
if item != None:
|
||||||
super(SyncMemStore, self).stop(item)
|
super(SyncRocksDbStore, self).stop(item)
|
||||||
logg.info('I am an in-memory only state store. I am shutting down now, so all state will now be discarded.')
|
logg.info('I am an in-memory only state store. I am shutting down now, so all state will now be discarded.')
|
||||||
|
|
||||||
|
|
||||||
def save_filter_list(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def load_filter_list(self):
|
|
||||||
return []
|
|
||||||
|
|
|
@ -43,8 +43,8 @@ class SyncRocksDbStore(SyncStore):
|
||||||
prefix_factory = RocksDbStoreAdder(self.factory, 'filter')
|
prefix_factory = RocksDbStoreAdder(self.factory, 'filter')
|
||||||
self.setup_filter_state(prefix_factory, filter_state_event_callback)
|
self.setup_filter_state(prefix_factory, filter_state_event_callback)
|
||||||
|
|
||||||
#self.session_id = os.path.basename(self.session_path)
|
self.session_id = os.path.basename(self.session_path)
|
||||||
#logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))
|
logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))
|
||||||
|
|
||||||
self.target_db = RocksDbStoreAdder(self.factory, '.stat').add('target')
|
self.target_db = RocksDbStoreAdder(self.factory, '.stat').add('target')
|
||||||
|
|
||||||
|
@ -64,16 +64,3 @@ class SyncRocksDbStore(SyncStore):
|
||||||
if item != None:
|
if item != None:
|
||||||
super(SyncRocksDbStore, self).stop(item)
|
super(SyncRocksDbStore, self).stop(item)
|
||||||
self.factory.close()
|
self.factory.close()
|
||||||
|
|
||||||
|
|
||||||
def save_filter_list(self):
|
|
||||||
fltr = []
|
|
||||||
for v in self.filters:
|
|
||||||
fltr.append(v.common_name())
|
|
||||||
self.target_db.put('filter_list', ','.join(fltr))
|
|
||||||
|
|
||||||
|
|
||||||
def load_filter_list(self):
|
|
||||||
v = self.target_db.get('filter_list')
|
|
||||||
v = v.decode('utf-8')
|
|
||||||
return v.split(',')
|
|
||||||
|
|
|
@ -5,7 +5,6 @@ import unittest
|
||||||
import shutil
|
import shutil
|
||||||
import tempfile
|
import tempfile
|
||||||
import logging
|
import logging
|
||||||
import uuid
|
|
||||||
|
|
||||||
# local imports
|
# local imports
|
||||||
from chainsyncer.session import SyncSession
|
from chainsyncer.session import SyncSession
|
||||||
|
@ -36,10 +35,7 @@ def filter_change_callback(k, old_state, new_state):
|
||||||
class TestStoreBase(unittest.TestCase):
|
class TestStoreBase(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.base_path = tempfile.mkdtemp()
|
self.path = tempfile.mkdtemp()
|
||||||
self.session_id = str(uuid.uuid4())
|
|
||||||
self.path = os.path.join(self.base_path, self.session_id)
|
|
||||||
os.makedirs(self.path)
|
|
||||||
self.store_factory = None
|
self.store_factory = None
|
||||||
self.persist = True
|
self.persist = True
|
||||||
|
|
||||||
|
@ -50,7 +46,6 @@ class TestStoreBase(unittest.TestCase):
|
||||||
"default",
|
"default",
|
||||||
"store_start",
|
"store_start",
|
||||||
"store_resume",
|
"store_resume",
|
||||||
"filter_list",
|
|
||||||
"sync_process_nofilter",
|
"sync_process_nofilter",
|
||||||
"sync_process_onefilter",
|
"sync_process_onefilter",
|
||||||
"sync_process_outoforder",
|
"sync_process_outoforder",
|
||||||
|
@ -71,25 +66,60 @@ class TestStoreBase(unittest.TestCase):
|
||||||
def t_default(self):
|
def t_default(self):
|
||||||
bogus_item = MockItem(0, 0, 0, 0)
|
bogus_item = MockItem(0, 0, 0, 0)
|
||||||
store = self.store_factory()
|
store = self.store_factory()
|
||||||
|
|
||||||
if store.session_path == None:
|
fp = os.path.join(self.path, store.session_id)
|
||||||
return
|
|
||||||
|
|
||||||
#fp = os.path.join(self.path, store.session_id)
|
|
||||||
fp = self.path
|
|
||||||
session_id = store.session_id
|
session_id = store.session_id
|
||||||
st = None
|
st = None
|
||||||
st = os.stat(fp)
|
try:
|
||||||
|
st = os.stat(fp)
|
||||||
|
except FileNotFoundError as e:
|
||||||
|
logg.warning('error {} persist {}'.format(e, self.persist))
|
||||||
|
if self.persist:
|
||||||
|
raise e
|
||||||
|
|
||||||
if st != None:
|
if st != None:
|
||||||
self.assertTrue(stat.S_ISDIR(st.st_mode))
|
self.assertTrue(stat.S_ISDIR(st.st_mode))
|
||||||
#self.assertTrue(store.is_default)
|
self.assertTrue(store.is_default)
|
||||||
|
|
||||||
|
fpd = os.path.join(self.path, 'default')
|
||||||
|
try:
|
||||||
|
st = os.stat(fpd)
|
||||||
|
except FileNotFoundError as e:
|
||||||
|
logg.warning('error {} persist {}'.format(e, self.persist))
|
||||||
|
if self.persist:
|
||||||
|
raise e
|
||||||
|
if st != None:
|
||||||
|
self.assertTrue(stat.S_ISDIR(st.st_mode))
|
||||||
|
self.assertTrue(store.is_default)
|
||||||
|
|
||||||
|
fpd = os.path.realpath(fpd)
|
||||||
|
self.assertEqual(fpd, fp)
|
||||||
|
|
||||||
store.stop(bogus_item)
|
store.stop(bogus_item)
|
||||||
store = self.store_factory()
|
store = self.store_factory()
|
||||||
fpr = os.path.join(self.path, self.session_id)
|
fpr = os.path.join(self.path, session_id)
|
||||||
self.assertEqual(fp, self.path)
|
self.assertEqual(fp, fpr)
|
||||||
|
self.assertTrue(store.is_default)
|
||||||
|
|
||||||
|
store.stop(bogus_item)
|
||||||
|
store = self.store_factory('default')
|
||||||
|
fpr = os.path.join(self.path, session_id)
|
||||||
|
self.assertEqual(fp, fpr)
|
||||||
|
self.assertTrue(store.is_default)
|
||||||
|
|
||||||
|
store.stop(bogus_item)
|
||||||
|
store = self.store_factory('foo')
|
||||||
|
fpf = os.path.join(self.path, 'foo')
|
||||||
|
try:
|
||||||
|
st = os.stat(fpf)
|
||||||
|
except FileNotFoundError as e:
|
||||||
|
logg.warning('error {} persist {}'.format(e, self.persist))
|
||||||
|
if self.persist:
|
||||||
|
raise e
|
||||||
|
if st != None:
|
||||||
|
self.assertTrue(stat.S_ISDIR(st.st_mode))
|
||||||
|
self.assertFalse(store.is_default)
|
||||||
|
|
||||||
|
|
||||||
def t_store_start(self):
|
def t_store_start(self):
|
||||||
bogus_item = MockItem(0, 0, 0, 0)
|
bogus_item = MockItem(0, 0, 0, 0)
|
||||||
|
@ -276,26 +306,3 @@ class TestStoreBase(unittest.TestCase):
|
||||||
o.next(advance_block=True)
|
o.next(advance_block=True)
|
||||||
with self.assertRaises(SyncDone):
|
with self.assertRaises(SyncDone):
|
||||||
o.next(advance_block=True)
|
o.next(advance_block=True)
|
||||||
|
|
||||||
|
|
||||||
def t_filter_list(self):
|
|
||||||
bogus_item = MockItem(0, 0, 0, 0)
|
|
||||||
store = self.store_factory()
|
|
||||||
|
|
||||||
if store.session_path == None:
|
|
||||||
return
|
|
||||||
|
|
||||||
fltr_one = MockFilter('foo_bar')
|
|
||||||
store.register(fltr_one)
|
|
||||||
|
|
||||||
fltr_two = MockFilter('bar_baz')
|
|
||||||
store.register(fltr_two)
|
|
||||||
|
|
||||||
store.start()
|
|
||||||
store.stop(bogus_item)
|
|
||||||
|
|
||||||
store = self.store_factory()
|
|
||||||
r = store.load_filter_list()
|
|
||||||
|
|
||||||
self.assertEqual(r[0], 'foo_bar')
|
|
||||||
self.assertEqual(r[1], 'bar_baz')
|
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
confini~=0.6.0
|
confini~=0.6.0
|
||||||
semver==2.13.0
|
semver==2.13.0
|
||||||
hexathon~=0.1.5
|
hexathon~=0.1.5
|
||||||
chainlib~=0.1.1
|
chainlib>=0.1.0b1,<0.2.0
|
||||||
shep~=0.2.3
|
shep~=0.2.3
|
||||||
|
|
10
setup.cfg
10
setup.cfg
|
@ -1,6 +1,6 @@
|
||||||
[metadata]
|
[metadata]
|
||||||
name = chainsyncer
|
name = chainsyncer
|
||||||
version = 0.4.1
|
version = 0.3.7
|
||||||
description = Generic blockchain syncer driver
|
description = Generic blockchain syncer driver
|
||||||
author = Louis Holbrook
|
author = Louis Holbrook
|
||||||
author_email = dev@holbrook.no
|
author_email = dev@holbrook.no
|
||||||
|
@ -29,13 +29,11 @@ packages =
|
||||||
chainsyncer.unittest
|
chainsyncer.unittest
|
||||||
chainsyncer.store
|
chainsyncer.store
|
||||||
chainsyncer.cli
|
chainsyncer.cli
|
||||||
chainsyncer.runnable
|
|
||||||
|
|
||||||
#[options.package_data]
|
#[options.package_data]
|
||||||
#* =
|
#* =
|
||||||
# sql/*
|
# sql/*
|
||||||
|
|
||||||
[options.entry_points]
|
#[options.entry_points]
|
||||||
console_scripts =
|
#console_scripts =
|
||||||
#blocksync-celery = chainsyncer.runnable.tracker:main
|
# blocksync-celery = chainsyncer.runnable.tracker:main
|
||||||
chainsyncer-unlock = chainsyncer.runnable.unlock:main
|
|
||||||
|
|
Loading…
Reference in New Issue