commit 384c79bed0cead2792767fb4ff2348017d84156f
parent 36acf3f09ac71d7a5e97abe25e2dec631e064e59
Author: lash <dev@holbrook.no>
Date: Thu, 28 Apr 2022 06:10:43 +0000
Remove session id path generation
Diffstat:
8 files changed, 46 insertions(+), 76 deletions(-)
diff --git a/chainsyncer/filter.py b/chainsyncer/filter.py
@@ -50,6 +50,7 @@ class FilterState:
self.unset = self.state_store.unset
self.peek = self.state_store.peek
self.from_name = self.state_store.from_name
+ self.list = self.state_store.list
self.state_store.sync()
self.all = self.state_store.all
self.started = False
@@ -93,6 +94,7 @@ class FilterState:
if self.scan != None:
ks = self.scan()
for v in ks: #os.listdir(self.scan_path):
+ logg.debug('ks {}'.format(v))
k = None
try:
k = self.state_store.from_elements(v)
diff --git a/chainsyncer/runnable/lock.py b/chainsyncer/runnable/lock.py
@@ -13,6 +13,7 @@ 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
logging.basicConfig(level=logging.WARNING)
@@ -37,6 +38,8 @@ logg.debug('config loaded:\n{}'.format(config))
settings = ChainsyncerSettings()
settings.process_sync_backend(config)
+logg.debug('settings:\n{}'.format(str(settings)))
+
def main():
@@ -51,13 +54,24 @@ def main():
syncer_store_class = getattr(syncer_store_module, 'RocksdbStoreFactory')
else:
raise NotImplementedError('cannot use backend: {}'.format(config.get('SYNCER_BACKEND')))
-
+
state_dir = config.get('_STATE_DIR')
factory = syncer_store_class(state_dir)
- base_state = PersistedState(factory.add, 0, check_alias=False)
- state = FilterState(base_state, scan=True)
- print(state)
+ store = SyncStore(state_dir, no_session=True)
+ #base_state = PersistedState(factory.add, 0, check_alias=False)
+ #state = FilterState(base_state, scan=True)
+ store.setup_filter_state(factory=factory)
+ store.connect()
+ store.filter_state.scan()
+ locked_state = store.filter_state.list(store.filter_state.from_name('RESET'))
+ print(locked_state)
+# if locked_state == None:
+# sys.stderr.write('state in {} backend "{}" is not locked\n'.format(state_dir, config.get('SYNCER_BACKEND')))
+# sys.exit(1)
+
+
+
if __name__ == '__main__':
main()
diff --git a/chainsyncer/settings.py b/chainsyncer/settings.py
@@ -6,16 +6,12 @@ from hexathon import (
to_int as hex_to_int,
strip_0x,
)
+from chainlib.settings import ChainSettings
logg = logging.getLogger(__name__)
-class ChainsyncerSettings:
-
- def __init__(self):
- self.o = {}
- self.get = self.o.get
-
+class ChainsyncerSettings(ChainSettings):
def process_sync_backend(self, config):
self.o['SYNCER_BACKEND'] = config.get('SYNCER_BACKEND')
diff --git a/chainsyncer/store/base.py b/chainsyncer/store/base.py
@@ -148,7 +148,7 @@ class SyncItem:
class SyncStore:
def __init__(self, path, session_id=None):
- self.session_id = None
+ self.session_id = session_id
self.session_path = None
self.is_default = False
self.first = False
@@ -157,16 +157,7 @@ class SyncStore:
self.item_keys = []
self.started = False
self.thresholds = []
- 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)
+ self.session_path = path
def setup_sync_state(self, factory=None, event_callback=None):
@@ -278,6 +269,7 @@ class SyncStore:
def get(self, k):
+ logg.debug('items {}'.format(self.items.keys()))
return self.items[k]
diff --git a/chainsyncer/store/mem.py b/chainsyncer/store/mem.py
@@ -14,10 +14,7 @@ logg = logging.getLogger(__name__)
class SyncMemStore(SyncStore):
def __init__(self, session_id=None, state_event_callback=None, filter_state_event_callback=None):
- 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))
+ super(SyncMemStore, self).__init__(None, session_id=session_id)
factory = None
self.setup_sync_state(factory, state_event_callback)
@@ -36,5 +33,5 @@ class SyncMemStore(SyncStore):
def stop(self, item):
if item != None:
- super(SyncRocksDbStore, self).stop(item)
+ super(SyncMemStore, 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.')
diff --git a/chainsyncer/store/rocksdb.py b/chainsyncer/store/rocksdb.py
@@ -43,8 +43,8 @@ class SyncRocksDbStore(SyncStore):
prefix_factory = RocksDbStoreAdder(self.factory, 'filter')
self.setup_filter_state(prefix_factory, filter_state_event_callback)
- self.session_id = os.path.basename(self.session_path)
- logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, 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))
self.target_db = RocksDbStoreAdder(self.factory, '.stat').add('target')
diff --git a/chainsyncer/unittest/store.py b/chainsyncer/unittest/store.py
@@ -5,6 +5,7 @@ import unittest
import shutil
import tempfile
import logging
+import uuid
# local imports
from chainsyncer.session import SyncSession
@@ -35,7 +36,10 @@ def filter_change_callback(k, old_state, new_state):
class TestStoreBase(unittest.TestCase):
def setUp(self):
- self.path = tempfile.mkdtemp()
+ self.base_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.persist = True
@@ -66,60 +70,25 @@ class TestStoreBase(unittest.TestCase):
def t_default(self):
bogus_item = MockItem(0, 0, 0, 0)
store = self.store_factory()
-
- fp = os.path.join(self.path, store.session_id)
+
+ if store.session_path == None:
+ return
+
+ #fp = os.path.join(self.path, store.session_id)
+ fp = self.path
session_id = store.session_id
st = None
- try:
- st = os.stat(fp)
- except FileNotFoundError as e:
- logg.warning('error {} persist {}'.format(e, self.persist))
- if self.persist:
- raise e
+ st = os.stat(fp)
if st != None:
self.assertTrue(stat.S_ISDIR(st.st_mode))
- 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)
+ #self.assertTrue(store.is_default)
store.stop(bogus_item)
store = self.store_factory()
- 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('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)
-
+ fpr = os.path.join(self.path, self.session_id)
+ self.assertEqual(fp, self.path)
+
def t_store_start(self):
bogus_item = MockItem(0, 0, 0, 0)
diff --git a/setup.cfg b/setup.cfg
@@ -1,6 +1,6 @@
[metadata]
name = chainsyncer
-version = 0.3.7
+version = 0.4.0
description = Generic blockchain syncer driver
author = Louis Holbrook
author_email = dev@holbrook.no
