Add one-shot channel implementation

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>

Author: shsms

src/frequenz/channels/__init__.py

@@ -92,6 +92,7 @@
 )
 from ._latest_value_cache import LatestValueCache
 from ._merge import Merger, merge
+from ._oneshot import OneshotChannel
 from ._receiver import Receiver, ReceiverError, ReceiverStoppedError
 from ._select import (
     Selected,
@@ -113,6 +114,7 @@
     "LatestValueCache",
     "MappedMessageT_co",
     "Merger",
+    "OneshotChannel",
     "Receiver",
     "ReceiverError",
     "ReceiverMessageT_co",

src/frequenz/channels/_oneshot.py

@@ -0,0 +1,142 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""A channel that can send a single message."""
+
+from __future__ import annotations
+
+import asyncio
+import typing
+
+from ._generic import ChannelMessageT
+from ._receiver import Receiver, ReceiverStoppedError
+from ._sender import Sender, SenderClosedError
+
+
+class _Empty:
+    """A sentinel indicating that no message has been sent."""
+
+
+_EMPTY = _Empty()
+
+
+class _Oneshot(typing.Generic[ChannelMessageT]):
+    """Internal representation of a one-shot channel.
+
+    A one-shot channel is a channel that can only send one message. After the first
+    message is sent, the sender is closed and any further attempts to send a message
+    will raise a `SenderClosedError`.
+
+    # Example
+
+    This example demonstrates how to use a one-shot channel to send a message
+    from one task to another.
+
+    ```python
+    import asyncio
+
+    from frequenz.channels import OneshotChannel, OneshotSender
+
+    async def send(sender: OneshotSender[int]) -> None:
+        await sender.send(42)
+
+    async def main() -> None:
+        sender, receiver = OneshotChannel[int]()
+
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(send(sender))
+            assert await receiver.receive() == 42
+
+    asyncio.run(main())
+    ```
+    """
+
+    def __init__(self) -> None:
+        """Create a new one-shot channel."""
+        self.message: ChannelMessageT | _Empty = _EMPTY
+        self.closed: bool = False
+        self.drained: bool = False
+        self.event: asyncio.Event = asyncio.Event()
+
+
+class OneshotSender(Sender[ChannelMessageT]):
+    """A sender for a one-shot channel."""
+
+    def __init__(self, channel: _Oneshot[ChannelMessageT]) -> None:
+        """Initialize this sender."""
+        self._channel = channel
+
+    async def send(self, message: ChannelMessageT, /) -> None:
+        """Send a message through this sender."""
+        if self._channel.closed:
+            raise SenderClosedError(self)
+        self._channel.message = message
+        self._channel.closed = True
+        self._channel.event.set()
+
+    async def aclose(self) -> None:
+        """Close this sender."""
+        self._channel.closed = True
+        if isinstance(self._channel.message, _Empty):
+            self._channel.drained = True
+        self._channel.event.set()
+
+
+class OneshotReceiver(Receiver[ChannelMessageT]):
+    """A receiver for a one-shot channel."""
+
+    def __init__(self, channel: _Oneshot[ChannelMessageT]) -> None:
+        """Initialize this receiver."""
+        self._channel = channel
+
+    async def ready(self) -> bool:
+        """Check if a message is ready to be received.
+
+        Returns:
+            `True` if a message is ready to be received, `False` if the sender
+                is closed and no message will be sent.
+        """
+        if self._channel.drained:
+            return False
+        while not self._channel.closed:
+            await self._channel.event.wait()
+        if isinstance(self._channel.message, _Empty):
+            return False
+        return True
+
+    def consume(self) -> ChannelMessageT:
+        """Consume a message from this receiver.
+
+        Returns:
+            The message that was sent through this channel.
+
+        Raises:
+            ReceiverStoppedError: If the sender was closed without sending a message.
+        """
+        if self._channel.drained:
+            raise ReceiverStoppedError(self)
+
+        assert not isinstance(
+            self._channel.message, _Empty
+        ), "`consume()` must be preceded by a call to `ready()`."
+
+        self._channel.drained = True
+        self._channel.event.clear()
+        return self._channel.message
+
+
+class OneshotChannel(
+    tuple[OneshotSender[ChannelMessageT], OneshotReceiver[ChannelMessageT]]
+):
+    """A one-shot channel.
+
+    A one-shot channel is a channel that can only send one message. After the first
+    message is sent, the sender is closed and any further attempts to send a message
+    will raise a `SenderClosedError`.
+    """
+
+    def __new__(cls) -> OneshotChannel[ChannelMessageT]:
+        """Create a new one-shot channel."""
+        channel = _Oneshot[ChannelMessageT]()
+
+        return tuple.__new__(cls, (OneshotSender(channel), OneshotReceiver(channel)))

tests/test_oneshot.py

@@ -0,0 +1,87 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the oneshot channel."""
+
+import asyncio
+
+import pytest
+
+from frequenz.channels import (
+    OneshotChannel,
+    ReceiverStoppedError,
+    SenderClosedError,
+)
+
+
+async def test_oneshot_recv_after_send() -> None:
+    """Test the oneshot function.
+
+    `receiver.receive()` is called after `sender.send()`.
+    """
+    sender, receiver = OneshotChannel[int]()
+
+    await sender.send(42)
+    assert await receiver.receive() == 42
+
+    with pytest.raises(SenderClosedError):
+        await sender.send(43)
+    with pytest.raises(ReceiverStoppedError):
+        await receiver.receive()
+
+
+async def test_oneshot_recv_before_send() -> None:
+    """Test the oneshot function.
+
+    `receiver.receive()` is called before `sender.send()`.
+    """
+    sender, receiver = OneshotChannel[int]()
+
+    task = asyncio.create_task(receiver.receive())
+
+    # Give the receiver a chance to start waiting
+    await asyncio.sleep(0.0)
+
+    await sender.send(42)
+    assert await task == 42
+
+    with pytest.raises(SenderClosedError):
+        await sender.send(43)
+    with pytest.raises(ReceiverStoppedError):
+        await receiver.receive()
+
+
+async def test_oneshot_recv_after_sender_closed() -> None:
+    """Test that closing sender works without sending a message.
+
+    `receiver.receive()` is called after `sender.aclose()`.
+    """
+    sender, receiver = OneshotChannel[int]()
+
+    await sender.aclose()
+
+    with pytest.raises(ReceiverStoppedError):
+        await receiver.receive()
+    with pytest.raises(SenderClosedError):
+        await sender.send(4)
+
+
+async def test_oneshot_recv_before_sender_closed() -> None:
+    """Test that closing sender works without sending a message.
+
+    `receiver.receive()` is called before `sender.aclose()`.
+    """
+    sender, receiver = OneshotChannel[int]()
+
+    task = asyncio.create_task(receiver.receive())
+
+    # Give the receiver a chance to start waiting
+    await asyncio.sleep(0.0)
+
+    await sender.aclose()
+
+    with pytest.raises(ReceiverStoppedError):
+        await task
+
+    with pytest.raises(SenderClosedError):
+        await sender.send(4)