channels

frequenz.channels

Frequenz Channels.

This package contains channel implementations.

Channels:

• Anycast: A channel that supports multiple senders and multiple receivers. A message sent through a sender will be received by exactly one receiver.

• Bidirectional: A channel providing a client and a service handle to send and receive bidirectionally.

• Broadcast: A channel to broadcast messages from multiple senders to multiple receivers. Each message sent through any of the senders is received by all of the receivers.

Other base classes:

• Peekable: An object to allow users to get a peek at the latest value in the channel, without consuming anything.

• Receiver: An object that can wait for and consume messages from a channel.

• Sender: An object that can send messages to a channel.

Utilities:

• util: A module with utilities, like special receivers that implement timers, file watchers, merge receivers, or wait for messages in multiple channels.

Exception classes:

• ChannelError: Base class for all errors related to channels.

• ChannelClosedError: Error raised when trying to operate (send, receive, etc.) through a closed channel.

Classes

frequenz.channels.Anycast

Bases: Generic[T]

A channel for sending data across async tasks.

Anycast channels support multiple senders and multiple receivers. A message sent through a sender will be received by exactly one receiver.

In cases where each message need to be received by every receiver, a Broadcast channel may be used.

Uses an deque internally, so Anycast channels are not thread-safe.

When there are multiple channel receivers, they can be awaited simultaneously using Select, Merge or MergeNamed.

Example

async def send(sender: channel.Sender) -> None:
    while True:
        next = random.randint(3, 17)
        print(f"sending: {next}")
        await sender.send(next)


async def recv(id: int, receiver: channel.Receiver) -> None:
    while True:
        next = await receiver.receive()
        print(f"receiver_{id} received {next}")
        await asyncio.sleep(0.1) # sleep (or work) with the data


acast = channel.Anycast()

sender = acast.new_sender()
receiver_1 = acast.new_receiver()

asyncio.create_task(send(sender))

await recv(1, receiver_1)

Check the tests and benchmarks directories for more examples.

Source code in frequenz/channels/_anycast.py

class Anycast(Generic[T]):
    """A channel for sending data across async tasks.

    Anycast channels support multiple senders and multiple receivers.  A message sent
    through a sender will be received by exactly one receiver.

    In cases where each message need to be received by every receiver, a
    [Broadcast][frequenz.channels.Broadcast] channel may be used.

    Uses an [deque][collections.deque] internally, so Anycast channels are not
    thread-safe.

    When there are multiple channel receivers, they can be awaited
    simultaneously using [Select][frequenz.channels.util.Select],
    [Merge][frequenz.channels.util.Merge] or
    [MergeNamed][frequenz.channels.util.MergeNamed].

    Example:
        ``` python
        async def send(sender: channel.Sender) -> None:
            while True:
                next = random.randint(3, 17)
                print(f"sending: {next}")
                await sender.send(next)


        async def recv(id: int, receiver: channel.Receiver) -> None:
            while True:
                next = await receiver.receive()
                print(f"receiver_{id} received {next}")
                await asyncio.sleep(0.1) # sleep (or work) with the data


        acast = channel.Anycast()

        sender = acast.new_sender()
        receiver_1 = acast.new_receiver()

        asyncio.create_task(send(sender))

        await recv(1, receiver_1)
        ```

        Check the `tests` and `benchmarks` directories for more examples.
    """

    def __init__(self, maxsize: int = 10) -> None:
        """Create an Anycast channel.

        Args:
            maxsize: Size of the channel's buffer.
        """
        self.limit: int = maxsize
        self.deque: Deque[T] = deque(maxlen=maxsize)
        self.send_cv: Condition = Condition()
        self.recv_cv: Condition = Condition()
        self.closed: bool = False

    async def close(self) -> None:
        """Close the channel.

        Any further attempts to [send()][frequenz.channels.Sender.send] data
        will return `False`.

        Receivers will still be able to drain the pending items on the channel,
        but after that, subsequent
        [receive()][frequenz.channels.Receiver.receive] calls will return `None`
        immediately.

        """
        self.closed = True
        async with self.send_cv:
            self.send_cv.notify_all()
        async with self.recv_cv:
            self.recv_cv.notify_all()

    def new_sender(self) -> Sender[T]:
        """Create a new sender.

        Returns:
            A Sender instance attached to the Anycast channel.
        """
        return Sender(self)

    def new_receiver(self) -> Receiver[T]:
        """Create a new receiver.

        Returns:
            A Receiver instance attached to the Anycast channel.
        """
        return Receiver(self)

Functions

__init__(maxsize=10)

Create an Anycast channel.

PARAMETER	DESCRIPTION
maxsize	Size of the channel's buffer. TYPE: int DEFAULT: 10

Source code in frequenz/channels/_anycast.py

def __init__(self, maxsize: int = 10) -> None:
    """Create an Anycast channel.

    Args:
        maxsize: Size of the channel's buffer.
    """
    self.limit: int = maxsize
    self.deque: Deque[T] = deque(maxlen=maxsize)
    self.send_cv: Condition = Condition()
    self.recv_cv: Condition = Condition()
    self.closed: bool = False

close() async

Close the channel.

Any further attempts to send() data will return False.

Receivers will still be able to drain the pending items on the channel, but after that, subsequent receive() calls will return None immediately.

Source code in frequenz/channels/_anycast.py

async def close(self) -> None:
    """Close the channel.

    Any further attempts to [send()][frequenz.channels.Sender.send] data
    will return `False`.

    Receivers will still be able to drain the pending items on the channel,
    but after that, subsequent
    [receive()][frequenz.channels.Receiver.receive] calls will return `None`
    immediately.

    """
    self.closed = True
    async with self.send_cv:
        self.send_cv.notify_all()
    async with self.recv_cv:
        self.recv_cv.notify_all()

new_receiver()

Create a new receiver.

RETURNS	DESCRIPTION
Receiver[T]	A Receiver instance attached to the Anycast channel.

Source code in frequenz/channels/_anycast.py

def new_receiver(self) -> Receiver[T]:
    """Create a new receiver.

    Returns:
        A Receiver instance attached to the Anycast channel.
    """
    return Receiver(self)

new_sender()

Create a new sender.

RETURNS	DESCRIPTION
Sender[T]	A Sender instance attached to the Anycast channel.

Source code in frequenz/channels/_anycast.py

def new_sender(self) -> Sender[T]:
    """Create a new sender.

    Returns:
        A Sender instance attached to the Anycast channel.
    """
    return Sender(self)

frequenz.channels.Bidirectional

Bases: Generic[T, U]

A wrapper class for simulating bidirectional channels.

Source code in frequenz/channels/_bidirectional.py

class Bidirectional(Generic[T, U]):
    """A wrapper class for simulating bidirectional channels."""

    class Handle(Sender[V], Receiver[W]):
        """A handle to a [Bidirectional][frequenz.channels.Bidirectional] instance.

        It can be used to send/receive values between the client and service.
        """

        def __init__(self, sender: Sender[V], receiver: Receiver[W]) -> None:
            """Create a `Bidirectional.Handle` instance.

            Args:
                sender: A sender to send values with.
                receiver: A receiver to receive values from.
            """
            self._sender = sender
            self._receiver = receiver

        async def send(self, msg: V) -> bool:
            """Send a value to the other side.

            Args:
                msg: The value to send.

            Returns:
                Whether the send was successful or not.
            """
            return await self._sender.send(msg)

        async def ready(self) -> None:
            """Wait until the receiver is ready with a value."""
            await self._receiver.ready()  # pylint: disable=protected-access

        def consume(self) -> W:
            """Return the latest value once `_ready` is complete.

            Returns:
                The next value that was received.
            """
            return self._receiver.consume()  # pylint: disable=protected-access

    def __init__(self, client_id: str, service_id: str) -> None:
        """Create a `Bidirectional` instance.

        Args:
            client_id: A name for the client, used to name the channels.
            service_id: A name for the service end of the channels.
        """
        self._client_id = client_id
        self._request_channel: Broadcast[T] = Broadcast(f"req_{service_id}_{client_id}")
        self._response_channel: Broadcast[U] = Broadcast(
            f"resp_{service_id}_{client_id}"
        )

        self._client_handle = Bidirectional.Handle(
            self._request_channel.new_sender(),
            self._response_channel.new_receiver(),
        )
        self._service_handle = Bidirectional.Handle(
            self._response_channel.new_sender(),
            self._request_channel.new_receiver(),
        )

    @property
    def client_handle(self) -> Bidirectional.Handle[T, U]:
        """Get a `Handle` for the client side to use.

        Returns:
            Object to send/receive messages with.
        """
        return self._client_handle

    @property
    def service_handle(self) -> Bidirectional.Handle[U, T]:
        """Get a `Handle` for the service side to use.

        Returns:
            Object to send/receive messages with.
        """
        return self._service_handle

Classes

Handle

Bases: Sender[V], Receiver[W]

A handle to a Bidirectional instance.

It can be used to send/receive values between the client and service.

Source code in frequenz/channels/_bidirectional.py

class Handle(Sender[V], Receiver[W]):
    """A handle to a [Bidirectional][frequenz.channels.Bidirectional] instance.

    It can be used to send/receive values between the client and service.
    """

    def __init__(self, sender: Sender[V], receiver: Receiver[W]) -> None:
        """Create a `Bidirectional.Handle` instance.

        Args:
            sender: A sender to send values with.
            receiver: A receiver to receive values from.
        """
        self._sender = sender
        self._receiver = receiver

    async def send(self, msg: V) -> bool:
        """Send a value to the other side.

        Args:
            msg: The value to send.

        Returns:
            Whether the send was successful or not.
        """
        return await self._sender.send(msg)

    async def ready(self) -> None:
        """Wait until the receiver is ready with a value."""
        await self._receiver.ready()  # pylint: disable=protected-access

    def consume(self) -> W:
        """Return the latest value once `_ready` is complete.

        Returns:
            The next value that was received.
        """
        return self._receiver.consume()  # pylint: disable=protected-access

Functions

__init__(sender, receiver)

Create a Bidirectional.Handle instance.

PARAMETER	DESCRIPTION
sender	A sender to send values with. TYPE: Sender[V]
receiver	A receiver to receive values from. TYPE: Receiver[W]

Source code in frequenz/channels/_bidirectional.py

def __init__(self, sender: Sender[V], receiver: Receiver[W]) -> None:
    """Create a `Bidirectional.Handle` instance.

    Args:
        sender: A sender to send values with.
        receiver: A receiver to receive values from.
    """
    self._sender = sender
    self._receiver = receiver

consume()

Return the latest value once _ready is complete.

RETURNS	DESCRIPTION
W	The next value that was received.

Source code in frequenz/channels/_bidirectional.py

def consume(self) -> W:
    """Return the latest value once `_ready` is complete.

    Returns:
        The next value that was received.
    """
    return self._receiver.consume()  # pylint: disable=protected-access

ready() async

Wait until the receiver is ready with a value.

Source code in frequenz/channels/_bidirectional.py

async def ready(self) -> None:
    """Wait until the receiver is ready with a value."""
    await self._receiver.ready()  # pylint: disable=protected-access

send(msg) async

Send a value to the other side.

PARAMETER	DESCRIPTION
msg	The value to send. TYPE: V

RETURNS	DESCRIPTION
bool	Whether the send was successful or not.

Source code in frequenz/channels/_bidirectional.py

async def send(self, msg: V) -> bool:
    """Send a value to the other side.

    Args:
        msg: The value to send.

    Returns:
        Whether the send was successful or not.
    """
    return await self._sender.send(msg)

Functions

__init__(client_id, service_id)

Create a Bidirectional instance.

PARAMETER	DESCRIPTION
client_id	A name for the client, used to name the channels. TYPE: str
service_id	A name for the service end of the channels. TYPE: str

Source code in frequenz/channels/_bidirectional.py

def __init__(self, client_id: str, service_id: str) -> None:
    """Create a `Bidirectional` instance.

    Args:
        client_id: A name for the client, used to name the channels.
        service_id: A name for the service end of the channels.
    """
    self._client_id = client_id
    self._request_channel: Broadcast[T] = Broadcast(f"req_{service_id}_{client_id}")
    self._response_channel: Broadcast[U] = Broadcast(
        f"resp_{service_id}_{client_id}"
    )

    self._client_handle = Bidirectional.Handle(
        self._request_channel.new_sender(),
        self._response_channel.new_receiver(),
    )
    self._service_handle = Bidirectional.Handle(
        self._response_channel.new_sender(),
        self._request_channel.new_receiver(),
    )

client_handle() property

Get a Handle for the client side to use.

RETURNS	DESCRIPTION
Bidirectional.Handle[T, U]	Object to send/receive messages with.

Source code in frequenz/channels/_bidirectional.py

@property
def client_handle(self) -> Bidirectional.Handle[T, U]:
    """Get a `Handle` for the client side to use.

    Returns:
        Object to send/receive messages with.
    """
    return self._client_handle

service_handle() property

Get a Handle for the service side to use.

RETURNS	DESCRIPTION
Bidirectional.Handle[U, T]	Object to send/receive messages with.

Source code in frequenz/channels/_bidirectional.py

@property
def service_handle(self) -> Bidirectional.Handle[U, T]:
    """Get a `Handle` for the service side to use.

    Returns:
        Object to send/receive messages with.
    """
    return self._service_handle

frequenz.channels.Broadcast

Bases: Generic[T]

A channel to broadcast messages to multiple receivers.

Broadcast channels can have multiple senders and multiple receivers. Each message sent through any of the senders is received by all of the receivers.

Internally, a broadcast receiver's buffer is implemented with just append/pop operations on either side of a deque, which are thread-safe. Because of this, Broadcast channels are thread-safe.

When there are multiple channel receivers, they can be awaited simultaneously using Select, Merge or MergeNamed.

Example

async def send(sender: channel.Sender) -> None:
    while True:
        next = random.randint(3, 17)
        print(f"sending: {next}")
        await sender.send(next)


async def recv(id: int, receiver: channel.Receiver) -> None:
    while True:
        next = await receiver.receive()
        print(f"receiver_{id} received {next}")
        await asyncio.sleep(0.1) # sleep (or work) with the data


bcast = channel.Broadcast()

sender = bcast.new_sender()
receiver_1 = bcast.new_receiver()

asyncio.create_task(send(sender))

await recv(1, receiver_1)

Check the tests and benchmarks directories for more examples.

Source code in frequenz/channels/_broadcast.py

class Broadcast(Generic[T]):
    """A channel to broadcast messages to multiple receivers.

    `Broadcast` channels can have multiple senders and multiple receivers. Each
    message sent through any of the senders is received by all of the
    receivers.

    Internally, a broadcast receiver's buffer is implemented with just
    append/pop operations on either side of a [deque][collections.deque], which
    are thread-safe.  Because of this, `Broadcast` channels are thread-safe.

    When there are multiple channel receivers, they can be awaited
    simultaneously using [Select][frequenz.channels.util.Select],
    [Merge][frequenz.channels.util.Merge] or
    [MergeNamed][frequenz.channels.util.MergeNamed].

    Example:
        ``` python
        async def send(sender: channel.Sender) -> None:
            while True:
                next = random.randint(3, 17)
                print(f"sending: {next}")
                await sender.send(next)


        async def recv(id: int, receiver: channel.Receiver) -> None:
            while True:
                next = await receiver.receive()
                print(f"receiver_{id} received {next}")
                await asyncio.sleep(0.1) # sleep (or work) with the data


        bcast = channel.Broadcast()

        sender = bcast.new_sender()
        receiver_1 = bcast.new_receiver()

        asyncio.create_task(send(sender))

        await recv(1, receiver_1)
        ```

        Check the `tests` and `benchmarks` directories for more examples.
    """

    def __init__(self, name: str, resend_latest: bool = False) -> None:
        """Create a Broadcast channel.

        Args:
            name: A name for the broadcast channel, typically based on the type
                of data sent through it.  Used to identify the channel in the
                logs.
            resend_latest: When True, every time a new receiver is created with
                `new_receiver`, it will automatically get sent the latest value
                on the channel.  This allows new receivers on slow streams to
                get the latest value as soon as they are created, without having
                to wait for the next message on the channel to arrive.
        """
        self.name: str = name
        self._resend_latest = resend_latest

        self.recv_cv: Condition = Condition()
        self.receivers: Dict[UUID, weakref.ReferenceType[Receiver[T]]] = {}
        self.closed: bool = False
        self._latest: Optional[T] = None

    async def close(self) -> None:
        """Close the Broadcast channel.

        Any further attempts to [send()][frequenz.channels.Sender.send] data
        will return `False`.

        Receivers will still be able to drain the pending items on their queues,
        but after that, subsequent
        [receive()][frequenz.channels.Receiver.receive] calls will return `None`
        immediately.
        """
        self._latest = None
        self.closed = True
        async with self.recv_cv:
            self.recv_cv.notify_all()

    def new_sender(self) -> Sender[T]:
        """Create a new broadcast sender.

        Returns:
            A Sender instance attached to the broadcast channel.
        """
        return Sender(self)

    def new_receiver(
        self, name: Optional[str] = None, maxsize: int = 50
    ) -> Receiver[T]:
        """Create a new broadcast receiver.

        Broadcast receivers have their own buffer, and when messages are not
        being consumed fast enough and the buffer fills up, old messages will
        get dropped just in this receiver.

        Args:
            name: A name to identify the receiver in the logs.
            maxsize: Size of the receiver's buffer.

        Returns:
            A Receiver instance attached to the broadcast channel.
        """
        uuid = uuid4()
        if name is None:
            name = str(uuid)
        recv: Receiver[T] = Receiver(uuid, name, maxsize, self)
        self.receivers[uuid] = weakref.ref(recv)
        if self._resend_latest and self._latest is not None:
            recv.enqueue(self._latest)
        return recv

    def new_peekable(self) -> Peekable[T]:
        """Create a new Peekable for the broadcast channel.

        A Peekable provides a [peek()][frequenz.channels.Peekable.peek] method
        that allows the user to get a peek at the latest value in the channel,
        without consuming anything.

        Returns:
            A Peekable to peek into the broadcast channel with.
        """
        return Peekable(self)

Functions

__init__(name, resend_latest=False)

Create a Broadcast channel.

PARAMETER	DESCRIPTION
name	A name for the broadcast channel, typically based on the type of data sent through it. Used to identify the channel in the logs. TYPE: str
resend_latest	When True, every time a new receiver is created with new_receiver, it will automatically get sent the latest value on the channel. This allows new receivers on slow streams to get the latest value as soon as they are created, without having to wait for the next message on the channel to arrive. TYPE: bool DEFAULT: False

Source code in frequenz/channels/_broadcast.py

def __init__(self, name: str, resend_latest: bool = False) -> None:
    """Create a Broadcast channel.

    Args:
        name: A name for the broadcast channel, typically based on the type
            of data sent through it.  Used to identify the channel in the
            logs.
        resend_latest: When True, every time a new receiver is created with
            `new_receiver`, it will automatically get sent the latest value
            on the channel.  This allows new receivers on slow streams to
            get the latest value as soon as they are created, without having
            to wait for the next message on the channel to arrive.
    """
    self.name: str = name
    self._resend_latest = resend_latest

    self.recv_cv: Condition = Condition()
    self.receivers: Dict[UUID, weakref.ReferenceType[Receiver[T]]] = {}
    self.closed: bool = False
    self._latest: Optional[T] = None

close() async

Close the Broadcast channel.

Any further attempts to send() data will return False.

Receivers will still be able to drain the pending items on their queues, but after that, subsequent receive() calls will return None immediately.

Source code in frequenz/channels/_broadcast.py

async def close(self) -> None:
    """Close the Broadcast channel.

    Any further attempts to [send()][frequenz.channels.Sender.send] data
    will return `False`.

    Receivers will still be able to drain the pending items on their queues,
    but after that, subsequent
    [receive()][frequenz.channels.Receiver.receive] calls will return `None`
    immediately.
    """
    self._latest = None
    self.closed = True
    async with self.recv_cv:
        self.recv_cv.notify_all()

new_peekable()

Create a new Peekable for the broadcast channel.

A Peekable provides a peek() method that allows the user to get a peek at the latest value in the channel, without consuming anything.

RETURNS	DESCRIPTION
Peekable[T]	A Peekable to peek into the broadcast channel with.

Source code in frequenz/channels/_broadcast.py

def new_peekable(self) -> Peekable[T]:
    """Create a new Peekable for the broadcast channel.

    A Peekable provides a [peek()][frequenz.channels.Peekable.peek] method
    that allows the user to get a peek at the latest value in the channel,
    without consuming anything.

    Returns:
        A Peekable to peek into the broadcast channel with.
    """
    return Peekable(self)

new_receiver(name=None, maxsize=50)

Create a new broadcast receiver.

Broadcast receivers have their own buffer, and when messages are not being consumed fast enough and the buffer fills up, old messages will get dropped just in this receiver.

PARAMETER	DESCRIPTION
name	A name to identify the receiver in the logs. TYPE: Optional[str] DEFAULT: None
maxsize	Size of the receiver's buffer. TYPE: int DEFAULT: 50

RETURNS	DESCRIPTION
Receiver[T]	A Receiver instance attached to the broadcast channel.

Source code in frequenz/channels/_broadcast.py

def new_receiver(
    self, name: Optional[str] = None, maxsize: int = 50
) -> Receiver[T]:
    """Create a new broadcast receiver.

    Broadcast receivers have their own buffer, and when messages are not
    being consumed fast enough and the buffer fills up, old messages will
    get dropped just in this receiver.

    Args:
        name: A name to identify the receiver in the logs.
        maxsize: Size of the receiver's buffer.

    Returns:
        A Receiver instance attached to the broadcast channel.
    """
    uuid = uuid4()
    if name is None:
        name = str(uuid)
    recv: Receiver[T] = Receiver(uuid, name, maxsize, self)
    self.receivers[uuid] = weakref.ref(recv)
    if self._resend_latest and self._latest is not None:
        recv.enqueue(self._latest)
    return recv

new_sender()

Create a new broadcast sender.

RETURNS	DESCRIPTION
Sender[T]	A Sender instance attached to the broadcast channel.

Source code in frequenz/channels/_broadcast.py

def new_sender(self) -> Sender[T]:
    """Create a new broadcast sender.

    Returns:
        A Sender instance attached to the broadcast channel.
    """
    return Sender(self)

frequenz.channels.ChannelClosedError

Bases: ChannelError

Error raised when trying to operate on a closed channel.

Source code in frequenz/channels/_base_classes.py

class ChannelClosedError(ChannelError):
    """Error raised when trying to operate on a closed channel."""

    def __init__(self, channel: Any = None):
        """Create a `ChannelClosedError` instance.

        Args:
            channel: A reference to the channel that was closed.
        """
        super().__init__(f"Channel {channel} was closed", channel)

Functions

__init__(channel=None)

Create a ChannelClosedError instance.

PARAMETER	DESCRIPTION
channel	A reference to the channel that was closed. TYPE: Any DEFAULT: None

Source code in frequenz/channels/_base_classes.py

def __init__(self, channel: Any = None):
    """Create a `ChannelClosedError` instance.

    Args:
        channel: A reference to the channel that was closed.
    """
    super().__init__(f"Channel {channel} was closed", channel)

frequenz.channels.ChannelError

Bases: RuntimeError

Base channel error.

All exceptions generated by channels inherit from this exception.

Source code in frequenz/channels/_base_classes.py

class ChannelError(RuntimeError):
    """Base channel error.

    All exceptions generated by channels inherit from this exception.
    """

    def __init__(self, message: Any, channel: Any = None):
        """Create a ChannelError instance.

        Args:
            message: An error message.
            channel: A reference to the channel that encountered the error.
        """
        super().__init__(message)
        self.channel: Any = channel

Functions

__init__(message, channel=None)

Create a ChannelError instance.

PARAMETER	DESCRIPTION
message	An error message. TYPE: Any
channel	A reference to the channel that encountered the error. TYPE: Any DEFAULT: None

Source code in frequenz/channels/_base_classes.py

def __init__(self, message: Any, channel: Any = None):
    """Create a ChannelError instance.

    Args:
        message: An error message.
        channel: A reference to the channel that encountered the error.
    """
    super().__init__(message)
    self.channel: Any = channel

frequenz.channels.Peekable

Bases: ABC, Generic[T]

A channel peekable.

A Peekable provides a peek() method that allows the user to get a peek at the latest value in the channel, without consuming anything.

Source code in frequenz/channels/_base_classes.py

class Peekable(ABC, Generic[T]):
    """A channel peekable.

    A Peekable provides a [peek()][frequenz.channels.Peekable] method that
    allows the user to get a peek at the latest value in the channel, without
    consuming anything.
    """

    @abstractmethod
    def peek(self) -> Optional[T]:
        """Return the latest value that was sent to the channel.

        Returns:
            The latest value received by the channel, and `None`, if nothing
                has been sent to the channel yet.
        """

Functions

peek() abstractmethod

Return the latest value that was sent to the channel.

RETURNS	DESCRIPTION
Optional[T]	The latest value received by the channel, and None, if nothing has been sent to the channel yet.

Source code in frequenz/channels/_base_classes.py

@abstractmethod
def peek(self) -> Optional[T]:
    """Return the latest value that was sent to the channel.

    Returns:
        The latest value received by the channel, and `None`, if nothing
            has been sent to the channel yet.
    """

frequenz.channels.Receiver

Bases: ABC, Generic[T]

A channel Receiver.

Source code in frequenz/channels/_base_classes.py

class Receiver(ABC, Generic[T]):
    """A channel Receiver."""

    async def __anext__(self) -> T:
        """Await the next value in the async iteration over received values.

        Returns:
            The next value received.

        Raises:
            StopAsyncIteration: if the underlying channel is closed.
        """
        try:
            await self.ready()
            return self.consume()
        except ChannelClosedError as exc:
            raise StopAsyncIteration() from exc

    @abstractmethod
    async def ready(self) -> None:
        """Wait until the receiver is ready with a value.

        Once a call to `ready()` has finished, the value should be read with a call to
        `consume()`.

        Raises:
            ChannelClosedError: if the underlying channel is closed.
        """

    @abstractmethod
    def consume(self) -> T:
        """Return the latest value once `ready()` is complete.

        `ready()` must be called before each call to `consume()`.

        Returns:
            The next value received.

        Raises:
            ChannelClosedError: if the underlying channel is closed.
        """

    def __aiter__(self) -> Receiver[T]:
        """Initialize the async iterator over received values.

        Returns:
            `self`, since no extra setup is needed for the iterator.
        """
        return self

    async def receive(self) -> T:
        """Receive a message from the channel.

        Raises:
            ChannelClosedError: if the underlying channel is closed.

        Returns:
            The received message.
        """
        try:
            received = await self.__anext__()  # pylint: disable=unnecessary-dunder-call
        except StopAsyncIteration as exc:
            raise ChannelClosedError() from exc
        return received

    def map(self, call: Callable[[T], U]) -> Receiver[U]:
        """Return a receiver with `call` applied on incoming messages.

        Args:
            call: function to apply on incoming messages.

        Returns:
            A `Receiver` to read results of the given function from.
        """
        return _Map(self, call)

    def into_peekable(self) -> Peekable[T]:
        """Convert the `Receiver` implementation into a `Peekable`.

        Once this function has been called, the receiver will no longer be
        usable, and calling `receive` on the receiver will raise an exception.

        Raises:
            NotImplementedError: when a `Receiver` implementation doesn't have
                a custom `into_peekable` implementation.
        """
        raise NotImplementedError("This receiver does not implement `into_peekable`")

Functions

__aiter__()

Initialize the async iterator over received values.

RETURNS	DESCRIPTION
Receiver[T]	self, since no extra setup is needed for the iterator.

Source code in frequenz/channels/_base_classes.py

def __aiter__(self) -> Receiver[T]:
    """Initialize the async iterator over received values.

    Returns:
        `self`, since no extra setup is needed for the iterator.
    """
    return self

__anext__() async

Await the next value in the async iteration over received values.

RETURNS	DESCRIPTION
T	The next value received.

RAISES	DESCRIPTION
StopAsyncIteration	if the underlying channel is closed.

Source code in frequenz/channels/_base_classes.py

async def __anext__(self) -> T:
    """Await the next value in the async iteration over received values.

    Returns:
        The next value received.

    Raises:
        StopAsyncIteration: if the underlying channel is closed.
    """
    try:
        await self.ready()
        return self.consume()
    except ChannelClosedError as exc:
        raise StopAsyncIteration() from exc

consume() abstractmethod

Return the latest value once ready() is complete.

ready() must be called before each call to consume().

RETURNS	DESCRIPTION
T	The next value received.

RAISES	DESCRIPTION
ChannelClosedError	if the underlying channel is closed.

Source code in frequenz/channels/_base_classes.py

@abstractmethod
def consume(self) -> T:
    """Return the latest value once `ready()` is complete.

    `ready()` must be called before each call to `consume()`.

    Returns:
        The next value received.

    Raises:
        ChannelClosedError: if the underlying channel is closed.
    """

into_peekable()

Convert the Receiver implementation into a Peekable.

Once this function has been called, the receiver will no longer be usable, and calling receive on the receiver will raise an exception.

RAISES	DESCRIPTION
NotImplementedError	when a Receiver implementation doesn't have a custom into_peekable implementation.

Source code in frequenz/channels/_base_classes.py

def into_peekable(self) -> Peekable[T]:
    """Convert the `Receiver` implementation into a `Peekable`.

    Once this function has been called, the receiver will no longer be
    usable, and calling `receive` on the receiver will raise an exception.

    Raises:
        NotImplementedError: when a `Receiver` implementation doesn't have
            a custom `into_peekable` implementation.
    """
    raise NotImplementedError("This receiver does not implement `into_peekable`")

map(call)

Return a receiver with call applied on incoming messages.

PARAMETER	DESCRIPTION
call	function to apply on incoming messages. TYPE: Callable[[T], U]

RETURNS	DESCRIPTION
Receiver[U]	A Receiver to read results of the given function from.

Source code in frequenz/channels/_base_classes.py

def map(self, call: Callable[[T], U]) -> Receiver[U]:
    """Return a receiver with `call` applied on incoming messages.

    Args:
        call: function to apply on incoming messages.

    Returns:
        A `Receiver` to read results of the given function from.
    """
    return _Map(self, call)

ready() abstractmethod async

Wait until the receiver is ready with a value.

Once a call to ready() has finished, the value should be read with a call to consume().

RAISES	DESCRIPTION
ChannelClosedError	if the underlying channel is closed.

Source code in frequenz/channels/_base_classes.py

@abstractmethod
async def ready(self) -> None:
    """Wait until the receiver is ready with a value.

    Once a call to `ready()` has finished, the value should be read with a call to
    `consume()`.

    Raises:
        ChannelClosedError: if the underlying channel is closed.
    """

receive() async

Receive a message from the channel.

RAISES	DESCRIPTION
ChannelClosedError	if the underlying channel is closed.

RETURNS	DESCRIPTION
T	The received message.

Source code in frequenz/channels/_base_classes.py

async def receive(self) -> T:
    """Receive a message from the channel.

    Raises:
        ChannelClosedError: if the underlying channel is closed.

    Returns:
        The received message.
    """
    try:
        received = await self.__anext__()  # pylint: disable=unnecessary-dunder-call
    except StopAsyncIteration as exc:
        raise ChannelClosedError() from exc
    return received

frequenz.channels.Sender

Bases: ABC, Generic[T]

A channel Sender.

Source code in frequenz/channels/_base_classes.py

class Sender(ABC, Generic[T]):
    """A channel Sender."""

    @abstractmethod
    async def send(self, msg: T) -> bool:
        """Send a message to the channel.

        Args:
            msg: The message to be sent.

        Returns:
            Whether the message was sent, based on whether the channel is open
                or not.
        """

Functions

send(msg) abstractmethod async

Send a message to the channel.

PARAMETER	DESCRIPTION
msg	The message to be sent. TYPE: T

RETURNS	DESCRIPTION
bool	Whether the message was sent, based on whether the channel is open or not.

Source code in frequenz/channels/_base_classes.py

@abstractmethod
async def send(self, msg: T) -> bool:
    """Send a message to the channel.

    Args:
        msg: The message to be sent.

    Returns:
        Whether the message was sent, based on whether the channel is open
            or not.
    """