Changed SequencedItemMapperForPopo to convert class to topic string.
Previously, notification logs from PopoApplications had event
classes rather than topic strings.
from typing import Tuple, Type, Dict, Any
from eventsourcing.infrastructure.sequenceditemmapper import SequencedItemMapper
+from eventsourcing.utils.topic import resolve_topic, get_topic
from eventsourcing.whitehead import TEvent
@@ -8,9 +9,9 @@ class SequencedItemMapperForPopo(SequencedItemMapper):
def get_event_class_and_attrs(
self, topic: str, state: bytes
) -> Tuple[Type[TEvent], Dict]:
- return topic, state # type: ignore
+ return resolve_topic(topic), state # type: ignore
def get_item_topic_and_state(
self, domain_event_class: type, event_attrs: Dict[str, Any]
) -> Tuple[str, bytes]:
- return domain_event_class, event_attrs # type: ignore
+ return get_topic(domain_event_class), event_attrs # type: ignore
