diff --git a/src/pipecat/utils/sync/__init__.py b/src/pipecat/utils/sync/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/utils/sync/base_notifier.py b/src/pipecat/utils/sync/base_notifier.py new file mode 100644 index 000000000..60321e282 --- /dev/null +++ b/src/pipecat/utils/sync/base_notifier.py @@ -0,0 +1,36 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Base notifier interface for Pipecat.""" + +from abc import ABC, abstractmethod + + +class BaseNotifier(ABC): + """Abstract base class for notification mechanisms. + + Provides a standard interface for implementing notification and waiting + patterns used for event coordination and signaling between components + in the Pipecat framework. + """ + + @abstractmethod + async def notify(self): + """Send a notification signal. + + Implementations should trigger any waiting coroutines or processes + that are blocked on this notifier. + """ + pass + + @abstractmethod + async def wait(self): + """Wait for a notification signal. + + Implementations should block until a notification is received + from the corresponding notify() call. + """ + pass diff --git a/src/pipecat/utils/sync/event_notifier.py b/src/pipecat/utils/sync/event_notifier.py new file mode 100644 index 000000000..0a553fb1e --- /dev/null +++ b/src/pipecat/utils/sync/event_notifier.py @@ -0,0 +1,45 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Event-based notifier implementation using asyncio Event primitives.""" + +import asyncio + +from pipecat.utils.sync.base_notifier import BaseNotifier + + +class EventNotifier(BaseNotifier): + """Event-based notifier using asyncio.Event for task synchronization. + + Provides a simple notification mechanism where one task can signal + an event and other tasks can wait for that event to occur. The event + is automatically cleared after each wait operation. + """ + + def __init__(self): + """Initialize the event notifier. + + Creates an internal asyncio.Event for managing notifications. + """ + self._event = asyncio.Event() + + async def notify(self): + """Signal the event to notify waiting tasks. + + Sets the internal event, causing any tasks waiting on this + notifier to be awakened. + """ + self._event.set() + + async def wait(self): + """Wait for the event to be signaled. + + Blocks until another task calls notify(). Automatically clears + the event after being awakened so subsequent calls will wait + for the next notification. + """ + await self._event.wait() + self._event.clear()