HeartbeatFunction handles None or an iterable of new values

Author: gwaramadze

quixstreams/core/stream/functions/heartbeat.py

@@ -24,9 +24,9 @@ def wrapper(
             timestamp: int,
             headers: Any,
         ):
-            if is_heartbeat_message(key, value):
-                # TODO: Heartbeats may return values (like expired windows)
-                func(timestamp)
+            if is_heartbeat_message(key, value) and (result := func(timestamp)):
+                for new_value, new_key, new_timestamp, new_headers in result:
+                    child_executor(new_value, new_key, new_timestamp, new_headers)
             child_executor(value, key, timestamp, headers)
 
         return wrapper

quixstreams/core/stream/functions/types.py

@@ -1,4 +1,4 @@
-from typing import Any, Callable, Iterable, Protocol, Tuple, Union
+from typing import Any, Callable, Iterable, Optional, Protocol, Tuple, Union
 
 __all__ = (
     "StreamCallback",
@@ -36,14 +36,7 @@ def __bool__(self) -> bool: ...
     [Any, Any, int, Any], Iterable[Tuple[Any, Any, int, Any]]
 ]
 
-HeartbeatCallback = Callable[
-    [int],  # timestamp
-    Union[
-        None,
-        Tuple[Any, Any, int, Any],  # single value
-        Iterable[Tuple[Any, Any, int, Any]],  # expanded values
-    ],
-]
+HeartbeatCallback = Callable[[int], Optional[Iterable[Tuple[Any, Any, int, Any]]]]
 
 StreamCallback = Union[
     ApplyCallback,