♻️ Refactor EventBus notify method

acapy_agent/core/event_bus.py

@@ -94,33 +94,39 @@ async def notify(self, profile: "Profile", event: Event):
             event (Event): event to emit
 
         """
-        # TODO don't block notifier until subscribers have all been called?
-        # TODO trigger each processor but don't await?
-        # TODO log errors but otherwise ignore?
-
-        LOGGER.debug("Notifying subscribers: %s", event)
-
-        partials = []
-        for pattern, subscribers in self.topic_patterns_to_subscribers.items():
-            match = pattern.match(event.topic)
-
-            if not match:
-                continue
-
-            for subscriber in subscribers:
-                partials.append(
-                    partial(
-                        subscriber,
-                        profile,
-                        event.with_metadata(EventMetadata(pattern, match)),
-                    )
-                )
+        partials = [
+            partial(
+                subscriber,
+                profile,
+                event.with_metadata(EventMetadata(pattern, match)),
+            )
+            for pattern, subscribers in self.topic_patterns_to_subscribers.items()
+            if (match := pattern.match(event.topic))
+            for subscriber in subscribers
+        ]
 
-        for processor in partials:
+        if not partials:
+            LOGGER.info("No subscribers for %s event", event.topic)
+            return
+
+        LOGGER.debug("Notifying %d subscribers for %s event", len(partials), event.topic)
+        for coro in partials:
             try:
-                await processor()
+                task = asyncio.create_task(coro)
+                task.add_done_callback(self._log_task_exception)
             except Exception:
-                LOGGER.exception("Error occurred while processing event")
+                LOGGER.exception(
+                    "Error occurred while scheduling %s event notification: %s",
+                    event.topic,
+                    coro,
+                )
+
+    def _log_task_exception(self, task: asyncio.Task):
+        """Log exceptions in background notification tasks."""
+        try:
+            task.result()
+        except Exception:
+            LOGGER.exception("Exception in notification background task")
 
     def subscribe(self, pattern: Pattern, processor: Callable):
         """Subscribe to an event.

acapy_agent/core/profile.py

@@ -123,6 +123,8 @@ async def notify(self, topic: str, payload: Any):
         event_bus = self.inject_or(EventBus)
         if event_bus:
             await event_bus.notify(self, Event(topic, payload))
+        else:
+            LOGGER.warning("No event bus found for profile %s", self.name)
 
     def __repr__(self) -> str:
         """Get a human readable string."""

acapy_agent/protocols/issue_credential/v2_0/models/cred_ex_record.py

@@ -207,7 +207,7 @@ async def emit_event(self, session: ProfileSession, payload: Optional[Any] = Non
             payload = V20CredExRecordWebhook(**payload)
             payload = payload.__dict__
 
-        await session.profile.notify(topic, payload)
+        await session.emit_event(topic, payload)
 
     @property
     def record_value(self) -> Mapping:

acapy_agent/protocols/present_proof/v2_0/models/pres_exchange.py

@@ -205,7 +205,7 @@ async def emit_event(self, session: ProfileSession, payload: Optional[Any] = Non
             payload = V20PresExRecordWebhook(**payload)
             payload = payload.__dict__
 
-        await session.profile.notify(topic, payload)
+        await session.emit_event(topic, payload)
 
     @property
     def record_value(self) -> Mapping: