chainsyncer

commit 63993cc223a0071616681efa352db4274cda66f4
parent 186f28a020862e461459b2504b54711ffb8a492d
Author: lash <dev@holbrook.no>
Date:   Sun,  6 Aug 2023 19:25:55 +0100

Add context field in filter invocations

Diffstat:
M
CHANGELOG
 | 
2
++
M
chainsyncer/driver/base.py
 | 
12
+++++++-----
M
chainsyncer/session.py
 | 
9
+++++++--
M
chainsyncer/unittest/base.py
 | 
2
+-
M
setup.cfg
 | 
2
+-

5 files changed, 18 insertions(+), 9 deletions(-)
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,3 +1,5 @@
+* 0.8.1
+	- Add context field to receive key-value collections from calling agent
 * 0.8.0
 	- Enable batch rpc retrievals of receipts
 * 0.7.1
diff --git a/chainsyncer/driver/base.py b/chainsyncer/driver/base.py
@@ -40,6 +40,7 @@ class SyncDriver:
         self.clock_id = time.CLOCK_MONOTONIC_RAW
         self.store.connect()
         self.store.start(offset=offset, target=target)
+        self.item = None
         if not SyncDriver.signal_set:
             for sig in SyncDriver.signal_request:
                 signal.signal(sig, self.__sig_terminate)
@@ -57,17 +58,18 @@ class SyncDriver:
         logg.info('termination requested!')
         SyncDriver.running_global = False
         self.running = False
+        self.session.stop(self.item)
 
 
-    def run(self, conn, interval=1):
+    def run(self, conn, interval=1, ctx=None):
         while self.running_global:
-            self.session = SyncSession(self.store)
-            item = self.session.start()
-            if item == None:
+            self.session = SyncSession(self.store, ctx=ctx)
+            self.item = self.session.start()
+            if self.item == None:
                 self.running = False
                 self.running_global = False
                 break
-            self.loop(conn, item, interval=interval)
+            self.loop(conn, self.item, interval=interval)
 
 
     def idle(self, interval):
diff --git a/chainsyncer/session.py b/chainsyncer/session.py
@@ -10,12 +10,13 @@ logg = logging.getLogger(__name__)
 
 class SyncSession:
 
-    def __init__(self, session_store):
+    def __init__(self, session_store, ctx=None):
         self.session_store = session_store
         self.started = self.session_store.started
         self.next = self.session_store.next_item
         self.item = None
         self.filters = self.session_store.filters
+        self.ctx = ctx
 
 
     def get(self, k):
@@ -30,6 +31,10 @@ class SyncSession:
 
     def stop(self, item):
         self.session_store.stop(item)
+        for fltr in self.filters:
+            stopper = getattr(fltr, 'stop', None)
+            if stopper != None:
+               stopper()
 
 
     def filter(self, conn, block, tx):
@@ -37,7 +42,7 @@ class SyncSession:
         for fltr in self.filters:
             logg.debug('executing filter {}'.format(fltr))
             self.item.advance()
-            interrupt = fltr.filter(conn, block, tx)
+            interrupt = fltr.filter(conn, block, tx, ctx=self.ctx)
             if not self.item.release(interrupt=interrupt):
                 break
         self.item.reset()
diff --git a/chainsyncer/unittest/base.py b/chainsyncer/unittest/base.py
@@ -157,7 +157,7 @@ class MockFilter:
         return self.name 
 
 
-    def filter(self, conn, block, tx):
+    def filter(self, conn, block, tx, ctx=None):
         r = False
         if self.brk_hard != None:
             r = True
diff --git a/setup.cfg b/setup.cfg
@@ -1,6 +1,6 @@
 [metadata]
 name = chainsyncer
-version = 0.8.0
+version = 0.8.1
 description = Generic blockchain syncer driver
 author = Louis Holbrook
 author_email = dev@holbrook.no