Adjusted collect_events() method, to support alternative aggregate and event types.
@@ -2,6 +2,7 @@ import os
from abc import ABC, abstractmethod
from copy import deepcopy
from dataclasses import dataclass
+from inspect import ismethod
from itertools import chain
from threading import Event, Lock
from typing import (
@@ -553,10 +554,11 @@ class ProcessingEvent:
self.aggregates[obj.id] = obj
for event in obj.collect_events():
self.events.append(event)
- elif hasattr(obj, "collect_events"):
+ elif obj is not None:
+ if hasattr(obj, "collect_events") and ismethod(obj.collect_events):
for event in obj.collect_events():
self.events.append(event)
- elif obj is not None:
+ else:
self.events.append(obj)
self.saved_kwargs.update(kwargs)
