utils(sync): move sync to utils.sync

This commit is contained in:
Aleix Conchillo Flaqué
2025-12-03 18:20:12 -08:00
parent 3ca94363ec
commit f37a53cc25
3 changed files with 81 additions and 0 deletions

View File

View File

@@ -0,0 +1,36 @@
#
# Copyright (c) 20242025, 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

View File

@@ -0,0 +1,45 @@
#
# Copyright (c) 20242025, 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()