Thread lock for policy-based event processing, so it's thread-safe.
from collections import OrderedDict, defaultdict
+from threading import Lock
from eventsourcing.application.pipeline import Pipeable
from eventsourcing.application.simple import Application
@@ -28,6 +29,7 @@ class ProcessApplication(Pipeable, Application):
self.readers = OrderedDict()
self.is_reader_position_ok = defaultdict(bool)
self._notification_generators = {}
+ self._policy_lock = Lock()
super(ProcessApplication, self).__init__(name=name, setup_table=setup_table, **kwargs)
subscribe(self.run, self.is_upstream_prompt)
@@ -86,6 +88,7 @@ class ProcessApplication(Pipeable, Application):
reader = self.readers[upstream_name]
while True:
+ with self._policy_lock:
# Get notification generator.
generator = self.get_notification_generator(upstream_name, advance_by)
try:
@@ -162,11 +165,11 @@ class ProcessApplication(Pipeable, Application):
except KeyError:
pass
raise exc
- else:
- # Publish a prompt if there are new notifications.
- # Todo: Optionally send events as prompts, saves pulling event if it arrives in order.
+
self.take_snapshots(new_events)
+ # Publish a prompt if there are new notifications.
+ # Todo: Optionally send events as prompts, saves pulling event if it arrives in order.
if any([event.__notifiable__ for event in new_events]):
self.publish_prompt()
