Skip to content

experimental

frequenz.channels.experimental ¤

Experimental channel primitives.

Warning

This package contains experimental channel primitives that are not yet considered stable. For more information on what to expect and how to use the experimental package please read the experimental package guidelines.

Classes¤

frequenz.channels.experimental.NopReceiver ¤

Bases: Receiver[ReceiverMessageT_co]

A place-holder receiver that will never receive a message.

Source code in frequenz/channels/experimental/_nop_receiver.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class NopReceiver(Receiver[ReceiverMessageT_co]):
    """A place-holder receiver that will never receive a message."""

    def __init__(self) -> None:
        """Initialize this instance."""
        self._closed: bool = False

    @override
    async def ready(self) -> bool:
        """Wait for ever unless the receiver is closed.

        Returns:
            Whether the receiver is still active.
        """
        if self._closed:
            return False
        await asyncio.Future()
        return False

    @override
    def consume(self) -> ReceiverMessageT_co:  # noqa: DOC503 (raised indirectly)
        """Raise `ReceiverError` unless the NopReceiver is closed.

        If the receiver is closed, then raise `ReceiverStoppedError`.

        Returns:
            The next message received.

        Raises:
            ReceiverStoppedError: If the receiver stopped producing messages.
            ReceiverError: If there is some problem with the underlying receiver.
        """
        if self._closed:
            raise ReceiverStoppedError(self)
        raise ReceiverError("`consume()` must be preceded by a call to `ready()`", self)

    @override
    def close(self) -> None:
        """Stop the receiver."""
        self._closed = True
Functions¤
__aiter__ ¤
__aiter__() -> Self

Get an async iterator over the received messages.

RETURNS DESCRIPTION
Self

This receiver, as it is already an async iterator.
Source code in frequenz/channels/_receiver.py 
242
243
244
245
246
247
248
def __aiter__(self) -> Self:
    """Get an async iterator over the received messages.

    Returns:
        This receiver, as it is already an async iterator.
    """
    return self
__anext__ async ¤
__anext__() -> ReceiverMessageT_co

Await the next message in the async iteration over received messages.

RETURNS DESCRIPTION
ReceiverMessageT_co

The next received message.
RAISES DESCRIPTION
StopAsyncIteration

If the receiver stopped producing messages.
ReceiverError

If there is some problem with the receiver.
Source code in frequenz/channels/_receiver.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
async def __anext__(self) -> ReceiverMessageT_co:  # noqa: DOC503
    """Await the next message in the async iteration over received messages.

    Returns:
        The next received message.

    Raises:
        StopAsyncIteration: If the receiver stopped producing messages.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        await self.ready()
        return self.consume()
    except ReceiverStoppedError as exc:
        raise StopAsyncIteration() from exc
__init__ ¤
__init__() -> None

Initialize this instance.

Source code in frequenz/channels/experimental/_nop_receiver.py 
21
22
23
def __init__(self) -> None:
    """Initialize this instance."""
    self._closed: bool = False
close ¤
close() -> None

Stop the receiver.

Source code in frequenz/channels/experimental/_nop_receiver.py 
54
55
56
57
@override
def close(self) -> None:
    """Stop the receiver."""
    self._closed = True
consume ¤
consume() -> ReceiverMessageT_co

Raise ReceiverError unless the NopReceiver is closed.

If the receiver is closed, then raise ReceiverStoppedError.

RETURNS DESCRIPTION
ReceiverMessageT_co

The next message received.
RAISES DESCRIPTION
ReceiverStoppedError

If the receiver stopped producing messages.
ReceiverError

If there is some problem with the underlying receiver.
Source code in frequenz/channels/experimental/_nop_receiver.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@override
def consume(self) -> ReceiverMessageT_co:  # noqa: DOC503 (raised indirectly)
    """Raise `ReceiverError` unless the NopReceiver is closed.

    If the receiver is closed, then raise `ReceiverStoppedError`.

    Returns:
        The next message received.

    Raises:
        ReceiverStoppedError: If the receiver stopped producing messages.
        ReceiverError: If there is some problem with the underlying receiver.
    """
    if self._closed:
        raise ReceiverStoppedError(self)
    raise ReceiverError("`consume()` must be preceded by a call to `ready()`", self)
filter ¤
filter(
    filter_function: Callable[
        [ReceiverMessageT_co],
        TypeGuard[FilteredMessageT_co],
    ],
) -> Receiver[FilteredMessageT_co]

filter(
    filter_function: Callable[[ReceiverMessageT_co], bool],
) -> Receiver[ReceiverMessageT_co]

filter(
    filter_function: (
        Callable[[ReceiverMessageT_co], bool]
        | Callable[
            [ReceiverMessageT_co],
            TypeGuard[FilteredMessageT_co],
        ]
    ),
) -> (
    Receiver[ReceiverMessageT_co]
    | Receiver[FilteredMessageT_co]
)

Apply a filter function on the messages on a receiver.

Note

You can pass a type guard as the filter function to narrow the type of the messages that pass the filter.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
filter_function

The function to be applied on incoming messages to determine if they should be received.

TYPE: Callable[[ReceiverMessageT_co], bool] | Callable[[ReceiverMessageT_co], TypeGuard[FilteredMessageT_co]]

RETURNS DESCRIPTION
Receiver[ReceiverMessageT_co] | Receiver[FilteredMessageT_co]

A new receiver that only receives messages that pass the filter.
Source code in frequenz/channels/_receiver.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
def filter(
    self,
    filter_function: (
        Callable[[ReceiverMessageT_co], bool]
        | Callable[[ReceiverMessageT_co], TypeGuard[FilteredMessageT_co]]
    ),
    /,
) -> Receiver[ReceiverMessageT_co] | Receiver[FilteredMessageT_co]:
    """Apply a filter function on the messages on a receiver.

    Note:
        You can pass a [type guard][typing.TypeGuard] as the filter function to
        narrow the type of the messages that pass the filter.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        filter_function: The function to be applied on incoming messages to
            determine if they should be received.

    Returns:
        A new receiver that only receives messages that pass the filter.
    """
    return _Filter(receiver=self, filter_function=filter_function)
map ¤
map(
    mapping_function: Callable[
        [ReceiverMessageT_co], MappedMessageT_co
    ],
) -> Receiver[MappedMessageT_co]

Apply a mapping function on the received message.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
mapping_function

The function to be applied on incoming messages.

TYPE: Callable[[ReceiverMessageT_co], MappedMessageT_co]

RETURNS DESCRIPTION
Receiver[MappedMessageT_co]

A new receiver that applies the function on the received messages.
Source code in frequenz/channels/_receiver.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def map(
    self, mapping_function: Callable[[ReceiverMessageT_co], MappedMessageT_co], /
) -> Receiver[MappedMessageT_co]:
    """Apply a mapping function on the received message.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        mapping_function: The function to be applied on incoming messages.

    Returns:
        A new receiver that applies the function on the received messages.
    """
    return _Mapper(receiver=self, mapping_function=mapping_function)
ready async ¤
ready() -> bool

Wait for ever unless the receiver is closed.

RETURNS DESCRIPTION
bool

Whether the receiver is still active.
Source code in frequenz/channels/experimental/_nop_receiver.py 
25
26
27
28
29
30
31
32
33
34
35
@override
async def ready(self) -> bool:
    """Wait for ever unless the receiver is closed.

    Returns:
        Whether the receiver is still active.
    """
    if self._closed:
        return False
    await asyncio.Future()
    return False
receive async ¤
receive() -> ReceiverMessageT_co

Receive a message.

RETURNS DESCRIPTION
ReceiverMessageT_co

The received message.
RAISES DESCRIPTION
ReceiverStoppedError

If there is some problem with the receiver.
ReceiverError

If there is some problem with the receiver.
Source code in frequenz/channels/_receiver.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def receive(self) -> ReceiverMessageT_co:  # noqa: DOC503
    """Receive a message.

    Returns:
        The received message.

    Raises:
        ReceiverStoppedError: If there is some problem with the receiver.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        received = await anext(self)
    except StopAsyncIteration as exc:
        # If we already had a cause and it was the receiver was stopped,
        # then reuse that error, as StopAsyncIteration is just an artifact
        # introduced by __anext__.
        if (
            isinstance(exc.__cause__, ReceiverStoppedError)
            and exc.__cause__.receiver is self
        ):
            # This is a false positive, we are actually checking __cause__ is a
            # ReceiverStoppedError which is an exception.
            raise exc.__cause__  # pylint: disable=raising-non-exception
        raise ReceiverStoppedError(self) from exc
    return received
triggered ¤
triggered(
    selected: Selected[Any],
) -> TypeGuard[Selected[ReceiverMessageT_co]]

Check whether this receiver was selected by select().

This method is used in conjunction with the Selected class to determine which receiver was selected in select() iteration.

It also works as a type guard to narrow the type of the Selected instance to the type of the receiver.

Please see select() for an example.

PARAMETER DESCRIPTION
selected

The result of a select() iteration.

TYPE: Selected[Any]

RETURNS DESCRIPTION
TypeGuard[Selected[ReceiverMessageT_co]]

Whether this receiver was selected.
Source code in frequenz/channels/_receiver.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def triggered(
    self, selected: Selected[Any]
) -> TypeGuard[Selected[ReceiverMessageT_co]]:
    """Check whether this receiver was selected by [`select()`][frequenz.channels.select].

    This method is used in conjunction with the
    [`Selected`][frequenz.channels.Selected] class to determine which receiver was
    selected in `select()` iteration.

    It also works as a [type guard][typing.TypeGuard] to narrow the type of the
    `Selected` instance to the type of the receiver.

    Please see [`select()`][frequenz.channels.select] for an example.

    Args:
        selected: The result of a `select()` iteration.

    Returns:
        Whether this receiver was selected.
    """
    if handled := selected._recv is self:  # pylint: disable=protected-access
        selected._handled = True  # pylint: disable=protected-access
    return handled

frequenz.channels.experimental.OptionalReceiver ¤

Bases: Receiver[ReceiverMessageT_co]

A receiver that will wait indefinitely if there is no underlying receiver.

This receiver is useful when the underlying receiver is not set initially. Instead of making if-else branches to check if the receiver is set, you can use this receiver to wait indefinitely if it is not set.

Source code in frequenz/channels/experimental/_optional_receiver.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@deprecated("Use `frequenz.channels.experimental.NopReceiver` instead.")
class OptionalReceiver(Receiver[ReceiverMessageT_co]):
    """A receiver that will wait indefinitely if there is no underlying receiver.

    This receiver is useful when the underlying receiver is not set initially.
    Instead of making `if-else` branches to check if the receiver is set, you can use
    this receiver to wait indefinitely if it is not set.
    """

    def __init__(self, receiver: Receiver[ReceiverMessageT_co] | None):
        """Initialize this instance.

        Args:
            receiver: The underlying receiver, or `None` if there is no receiver.
        """
        self._receiver: Receiver[ReceiverMessageT_co] | None = receiver

    @override
    async def ready(self) -> bool:
        """Wait until the receiver is ready with a message or an error.

        Once a call to `ready()` has finished, the message should be read with
        a call to `consume()` (`receive()` or iterated over). The receiver will
        remain ready (this method will return immediately) until it is
        consumed.

        Returns:
            Whether the receiver is still active.
        """
        if self._receiver is not None:
            return await self._receiver.ready()

        # If there's no receiver, wait forever
        await asyncio.Event().wait()
        return False

    @override
    def consume(self) -> ReceiverMessageT_co:  # noqa: DOC503 (raised indirectly)
        """Return the latest from the underlying receiver message once `ready()` is complete.

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

        Returns:
            The next message received.

        Raises:
            ReceiverStoppedError: If the receiver stopped producing messages.
            ReceiverError: If there is some problem with the underlying receiver.
        """
        if self._receiver is None:
            raise ReceiverError(
                "`consume()` must be preceded by a call to `ready()`", self
            )
        return self._receiver.consume()

    def is_set(self) -> bool:
        """Check if the receiver is set."""
        return self._receiver is not None

    def close(self) -> None:
        """Stop the receiver."""
        if self._receiver is not None:
            self._receiver.close()
Functions¤
__aiter__ ¤
__aiter__() -> Self

Get an async iterator over the received messages.

RETURNS DESCRIPTION
Self

This receiver, as it is already an async iterator.
Source code in frequenz/channels/_receiver.py 
242
243
244
245
246
247
248
def __aiter__(self) -> Self:
    """Get an async iterator over the received messages.

    Returns:
        This receiver, as it is already an async iterator.
    """
    return self
__anext__ async ¤
__anext__() -> ReceiverMessageT_co

Await the next message in the async iteration over received messages.

RETURNS DESCRIPTION
ReceiverMessageT_co

The next received message.
RAISES DESCRIPTION
StopAsyncIteration

If the receiver stopped producing messages.
ReceiverError

If there is some problem with the receiver.
Source code in frequenz/channels/_receiver.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
async def __anext__(self) -> ReceiverMessageT_co:  # noqa: DOC503
    """Await the next message in the async iteration over received messages.

    Returns:
        The next received message.

    Raises:
        StopAsyncIteration: If the receiver stopped producing messages.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        await self.ready()
        return self.consume()
    except ReceiverStoppedError as exc:
        raise StopAsyncIteration() from exc
__init__ ¤
__init__(receiver: Receiver[ReceiverMessageT_co] | None)

Initialize this instance.

PARAMETER DESCRIPTION
receiver

The underlying receiver, or None if there is no receiver.

TYPE: Receiver[ReceiverMessageT_co] | None

Source code in frequenz/channels/experimental/_optional_receiver.py 
27
28
29
30
31
32
33
def __init__(self, receiver: Receiver[ReceiverMessageT_co] | None):
    """Initialize this instance.

    Args:
        receiver: The underlying receiver, or `None` if there is no receiver.
    """
    self._receiver: Receiver[ReceiverMessageT_co] | None = receiver
close ¤
close() -> None

Stop the receiver.

Source code in frequenz/channels/experimental/_optional_receiver.py 
77
78
79
80
def close(self) -> None:
    """Stop the receiver."""
    if self._receiver is not None:
        self._receiver.close()
consume ¤
consume() -> ReceiverMessageT_co

Return the latest from the underlying receiver message once ready() is complete.

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

RETURNS DESCRIPTION
ReceiverMessageT_co

The next message received.
RAISES DESCRIPTION
ReceiverStoppedError

If the receiver stopped producing messages.
ReceiverError

If there is some problem with the underlying receiver.
Source code in frequenz/channels/experimental/_optional_receiver.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@override
def consume(self) -> ReceiverMessageT_co:  # noqa: DOC503 (raised indirectly)
    """Return the latest from the underlying receiver message once `ready()` is complete.

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

    Returns:
        The next message received.

    Raises:
        ReceiverStoppedError: If the receiver stopped producing messages.
        ReceiverError: If there is some problem with the underlying receiver.
    """
    if self._receiver is None:
        raise ReceiverError(
            "`consume()` must be preceded by a call to `ready()`", self
        )
    return self._receiver.consume()
filter ¤
filter(
    filter_function: Callable[
        [ReceiverMessageT_co],
        TypeGuard[FilteredMessageT_co],
    ],
) -> Receiver[FilteredMessageT_co]

filter(
    filter_function: Callable[[ReceiverMessageT_co], bool],
) -> Receiver[ReceiverMessageT_co]

filter(
    filter_function: (
        Callable[[ReceiverMessageT_co], bool]
        | Callable[
            [ReceiverMessageT_co],
            TypeGuard[FilteredMessageT_co],
        ]
    ),
) -> (
    Receiver[ReceiverMessageT_co]
    | Receiver[FilteredMessageT_co]
)

Apply a filter function on the messages on a receiver.

Note

You can pass a type guard as the filter function to narrow the type of the messages that pass the filter.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
filter_function

The function to be applied on incoming messages to determine if they should be received.

TYPE: Callable[[ReceiverMessageT_co], bool] | Callable[[ReceiverMessageT_co], TypeGuard[FilteredMessageT_co]]

RETURNS DESCRIPTION
Receiver[ReceiverMessageT_co] | Receiver[FilteredMessageT_co]

A new receiver that only receives messages that pass the filter.
Source code in frequenz/channels/_receiver.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
def filter(
    self,
    filter_function: (
        Callable[[ReceiverMessageT_co], bool]
        | Callable[[ReceiverMessageT_co], TypeGuard[FilteredMessageT_co]]
    ),
    /,
) -> Receiver[ReceiverMessageT_co] | Receiver[FilteredMessageT_co]:
    """Apply a filter function on the messages on a receiver.

    Note:
        You can pass a [type guard][typing.TypeGuard] as the filter function to
        narrow the type of the messages that pass the filter.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        filter_function: The function to be applied on incoming messages to
            determine if they should be received.

    Returns:
        A new receiver that only receives messages that pass the filter.
    """
    return _Filter(receiver=self, filter_function=filter_function)
is_set ¤
is_set() -> bool

Check if the receiver is set.

Source code in frequenz/channels/experimental/_optional_receiver.py 
73
74
75
def is_set(self) -> bool:
    """Check if the receiver is set."""
    return self._receiver is not None
map ¤
map(
    mapping_function: Callable[
        [ReceiverMessageT_co], MappedMessageT_co
    ],
) -> Receiver[MappedMessageT_co]

Apply a mapping function on the received message.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
mapping_function

The function to be applied on incoming messages.

TYPE: Callable[[ReceiverMessageT_co], MappedMessageT_co]

RETURNS DESCRIPTION
Receiver[MappedMessageT_co]

A new receiver that applies the function on the received messages.
Source code in frequenz/channels/_receiver.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def map(
    self, mapping_function: Callable[[ReceiverMessageT_co], MappedMessageT_co], /
) -> Receiver[MappedMessageT_co]:
    """Apply a mapping function on the received message.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        mapping_function: The function to be applied on incoming messages.

    Returns:
        A new receiver that applies the function on the received messages.
    """
    return _Mapper(receiver=self, mapping_function=mapping_function)
ready async ¤
ready() -> bool

Wait until the receiver is ready with a message or an error.

Once a call to ready() has finished, the message should be read with a call to consume() (receive() or iterated over). The receiver will remain ready (this method will return immediately) until it is consumed.

RETURNS DESCRIPTION
bool

Whether the receiver is still active.
Source code in frequenz/channels/experimental/_optional_receiver.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@override
async def ready(self) -> bool:
    """Wait until the receiver is ready with a message or an error.

    Once a call to `ready()` has finished, the message should be read with
    a call to `consume()` (`receive()` or iterated over). The receiver will
    remain ready (this method will return immediately) until it is
    consumed.

    Returns:
        Whether the receiver is still active.
    """
    if self._receiver is not None:
        return await self._receiver.ready()

    # If there's no receiver, wait forever
    await asyncio.Event().wait()
    return False
receive async ¤
receive() -> ReceiverMessageT_co

Receive a message.

RETURNS DESCRIPTION
ReceiverMessageT_co

The received message.
RAISES DESCRIPTION
ReceiverStoppedError

If there is some problem with the receiver.
ReceiverError

If there is some problem with the receiver.
Source code in frequenz/channels/_receiver.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def receive(self) -> ReceiverMessageT_co:  # noqa: DOC503
    """Receive a message.

    Returns:
        The received message.

    Raises:
        ReceiverStoppedError: If there is some problem with the receiver.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        received = await anext(self)
    except StopAsyncIteration as exc:
        # If we already had a cause and it was the receiver was stopped,
        # then reuse that error, as StopAsyncIteration is just an artifact
        # introduced by __anext__.
        if (
            isinstance(exc.__cause__, ReceiverStoppedError)
            and exc.__cause__.receiver is self
        ):
            # This is a false positive, we are actually checking __cause__ is a
            # ReceiverStoppedError which is an exception.
            raise exc.__cause__  # pylint: disable=raising-non-exception
        raise ReceiverStoppedError(self) from exc
    return received
triggered ¤
triggered(
    selected: Selected[Any],
) -> TypeGuard[Selected[ReceiverMessageT_co]]

Check whether this receiver was selected by select().

This method is used in conjunction with the Selected class to determine which receiver was selected in select() iteration.

It also works as a type guard to narrow the type of the Selected instance to the type of the receiver.

Please see select() for an example.

PARAMETER DESCRIPTION
selected

The result of a select() iteration.

TYPE: Selected[Any]

RETURNS DESCRIPTION
TypeGuard[Selected[ReceiverMessageT_co]]

Whether this receiver was selected.
Source code in frequenz/channels/_receiver.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def triggered(
    self, selected: Selected[Any]
) -> TypeGuard[Selected[ReceiverMessageT_co]]:
    """Check whether this receiver was selected by [`select()`][frequenz.channels.select].

    This method is used in conjunction with the
    [`Selected`][frequenz.channels.Selected] class to determine which receiver was
    selected in `select()` iteration.

    It also works as a [type guard][typing.TypeGuard] to narrow the type of the
    `Selected` instance to the type of the receiver.

    Please see [`select()`][frequenz.channels.select] for an example.

    Args:
        selected: The result of a `select()` iteration.

    Returns:
        Whether this receiver was selected.
    """
    if handled := selected._recv is self:  # pylint: disable=protected-access
        selected._handled = True  # pylint: disable=protected-access
    return handled

frequenz.channels.experimental.Pipe ¤

Bases: Generic[ChannelMessageT]

A pipe between two channels.

The Pipe class takes a receiver and a sender and creates a pipe between them by forwarding all the messages received by the receiver to the sender.

Example 
import asyncio
from contextlib import closing, aclosing, AsyncExitStack

from frequenz.channels import Broadcast, Pipe, Receiver

async def main() -> None:
    # Channels, receivers and Pipe are in AsyncExitStack
    # to close and stop them at the end.
    async with AsyncExitStack() as stack:
        source_channel = await stack.enter_async_context(
            aclosing(Broadcast[int](name="source channel"))
        )
        source_receiver = stack.enter_context(closing(source_channel.new_receiver()))

        forwarding_channel = await stack.enter_async_context(
            aclosing(Broadcast[int](name="forwarding channel"))
        )
        await stack.enter_async_context(
            Pipe(source_receiver, forwarding_channel.new_sender())
        )

        receiver = stack.enter_context(closing(forwarding_channel.new_receiver()))

        source_sender = source_channel.new_sender()
        await source_sender.send(10)
        assert await receiver.receive() == 11

asyncio.run(main())
Source code in frequenz/channels/experimental/_pipe.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class Pipe(typing.Generic[ChannelMessageT]):
    """A pipe between two channels.

    The `Pipe` class takes a receiver and a sender and creates a pipe between them
    by forwarding all the messages received by the receiver to the sender.

    Example:
        ```python
        import asyncio
        from contextlib import closing, aclosing, AsyncExitStack

        from frequenz.channels import Broadcast, Pipe, Receiver

        async def main() -> None:
            # Channels, receivers and Pipe are in AsyncExitStack
            # to close and stop them at the end.
            async with AsyncExitStack() as stack:
                source_channel = await stack.enter_async_context(
                    aclosing(Broadcast[int](name="source channel"))
                )
                source_receiver = stack.enter_context(closing(source_channel.new_receiver()))

                forwarding_channel = await stack.enter_async_context(
                    aclosing(Broadcast[int](name="forwarding channel"))
                )
                await stack.enter_async_context(
                    Pipe(source_receiver, forwarding_channel.new_sender())
                )

                receiver = stack.enter_context(closing(forwarding_channel.new_receiver()))

                source_sender = source_channel.new_sender()
                await source_sender.send(10)
                assert await receiver.receive() == 11

        asyncio.run(main())
        ```
    """

    def __init__(
        self, receiver: Receiver[ChannelMessageT], sender: Sender[ChannelMessageT]
    ) -> None:
        """Create a new pipe between two channels.

        Args:
            receiver: The receiver channel.
            sender: The sender channel.
        """
        self._sender = sender
        self._receiver = receiver
        self._task: asyncio.Task[None] | None = None

    async def __aenter__(self) -> Pipe[ChannelMessageT]:
        """Enter the runtime context."""
        await self.start()
        return self

    async def __aexit__(
        self,
        _exc_type: typing.Type[BaseException] | None,
        _exc: BaseException | None,
        _tb: typing.Any,
    ) -> None:
        """Exit the runtime context."""
        await self.stop()

    async def start(self) -> None:
        """Start this pipe if it is not already running."""
        if not self._task or self._task.done():
            self._task = asyncio.create_task(self._run())

    async def stop(self) -> None:
        """Stop this pipe."""
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    async def _run(self) -> None:
        async for value in self._receiver:
            await self._sender.send(value)
Functions¤
__aenter__ async ¤
__aenter__() -> Pipe[ChannelMessageT]

Enter the runtime context.

Source code in frequenz/channels/experimental/_pipe.py 
72
73
74
75
async def __aenter__(self) -> Pipe[ChannelMessageT]:
    """Enter the runtime context."""
    await self.start()
    return self
__aexit__ async ¤
__aexit__(
    _exc_type: Type[BaseException] | None,
    _exc: BaseException | None,
    _tb: Any,
) -> None

Exit the runtime context.

Source code in frequenz/channels/experimental/_pipe.py 
77
78
79
80
81
82
83
84
async def __aexit__(
    self,
    _exc_type: typing.Type[BaseException] | None,
    _exc: BaseException | None,
    _tb: typing.Any,
) -> None:
    """Exit the runtime context."""
    await self.stop()
__init__ ¤
__init__(
    receiver: Receiver[ChannelMessageT],
    sender: Sender[ChannelMessageT],
) -> None

Create a new pipe between two channels.

PARAMETER DESCRIPTION
receiver

The receiver channel.

TYPE: Receiver[ChannelMessageT]

sender

The sender channel.

TYPE: Sender[ChannelMessageT]

Source code in frequenz/channels/experimental/_pipe.py 
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self, receiver: Receiver[ChannelMessageT], sender: Sender[ChannelMessageT]
) -> None:
    """Create a new pipe between two channels.

    Args:
        receiver: The receiver channel.
        sender: The sender channel.
    """
    self._sender = sender
    self._receiver = receiver
    self._task: asyncio.Task[None] | None = None
start async ¤
start() -> None

Start this pipe if it is not already running.

Source code in frequenz/channels/experimental/_pipe.py 
86
87
88
89
async def start(self) -> None:
    """Start this pipe if it is not already running."""
    if not self._task or self._task.done():
        self._task = asyncio.create_task(self._run())
stop async ¤
stop() -> None

Stop this pipe.

Source code in frequenz/channels/experimental/_pipe.py 
91
92
93
94
95
96
97
98
async def stop(self) -> None:
    """Stop this pipe."""
    if self._task and not self._task.done():
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass

frequenz.channels.experimental.RelaySender ¤

Bases: Generic[SenderMessageT_contra], Sender[SenderMessageT_contra]

A Sender for sending messages to multiple senders.

The RelaySender class takes multiple senders and forwards all the messages sent to it, to the senders it was created with.

Example 
from frequenz.channels import Broadcast
from frequenz.channels.experimental import RelaySender

channel1: Broadcast[int] = Broadcast(name="channel1")
channel2: Broadcast[int] = Broadcast(name="channel2")

receiver1 = channel1.new_receiver()
receiver2 = channel2.new_receiver()

tee_sender = RelaySender(channel1.new_sender(), channel2.new_sender())

await tee_sender.send(5)
assert await receiver1.receive() == 5
assert await receiver2.receive() == 5
Source code in frequenz/channels/experimental/_relay_sender.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class RelaySender(typing.Generic[SenderMessageT_contra], Sender[SenderMessageT_contra]):
    """A Sender for sending messages to multiple senders.

    The `RelaySender` class takes multiple senders and forwards all the messages sent to
    it, to the senders it was created with.

    Example:
        ```python
        from frequenz.channels import Broadcast
        from frequenz.channels.experimental import RelaySender

        channel1: Broadcast[int] = Broadcast(name="channel1")
        channel2: Broadcast[int] = Broadcast(name="channel2")

        receiver1 = channel1.new_receiver()
        receiver2 = channel2.new_receiver()

        tee_sender = RelaySender(channel1.new_sender(), channel2.new_sender())

        await tee_sender.send(5)
        assert await receiver1.receive() == 5
        assert await receiver2.receive() == 5
        ```
    """

    def __init__(self, *senders: Sender[SenderMessageT_contra]) -> None:
        """Create a new RelaySender.

        Args:
            *senders: The senders to send messages to.
        """
        self._senders = senders

    @override
    async def send(self, message: SenderMessageT_contra, /) -> None:
        """Send a message.

        Args:
            message: The message to be sent.
        """
        for sender in self._senders:
            await sender.send(message)
Functions¤
__init__ ¤
__init__(*senders: Sender[SenderMessageT_contra]) -> None

Create a new RelaySender.

PARAMETER DESCRIPTION
*senders

The senders to send messages to.

TYPE: Sender[SenderMessageT_contra] DEFAULT: ()

Source code in frequenz/channels/experimental/_relay_sender.py 
43
44
45
46
47
48
49
def __init__(self, *senders: Sender[SenderMessageT_contra]) -> None:
    """Create a new RelaySender.

    Args:
        *senders: The senders to send messages to.
    """
    self._senders = senders
send async ¤
send(message: SenderMessageT_contra) -> None

Send a message.

PARAMETER DESCRIPTION
message

The message to be sent.

TYPE: SenderMessageT_contra

Source code in frequenz/channels/experimental/_relay_sender.py 
51
52
53
54
55
56
57
58
59
@override
async def send(self, message: SenderMessageT_contra, /) -> None:
    """Send a message.

    Args:
        message: The message to be sent.
    """
    for sender in self._senders:
        await sender.send(message)

frequenz.channels.experimental.WithPrevious ¤

Bases: Generic[ChannelMessageT]

A composable predicate to build predicates that can use also the previous message.

This predicate can be used to filter messages based on a custom condition on the previous and current messages. This can be useful in cases where you want to process messages only if they satisfy a particular condition with respect to the previous message.

Receiving only messages that are different from the previous one. 
from frequenz.channels import Broadcast
from frequenz.channels.experimental import WithPrevious

channel = Broadcast[int](name="example")
receiver = channel.new_receiver().filter(WithPrevious(lambda old, new: old != new))
sender = channel.new_sender()

# This message will be received as it is the first message.
await sender.send(1)
assert await receiver.receive() == 1

# This message will be skipped as it equals to the previous one.
await sender.send(1)

# This message will be received as it is a different from the previous one.
await sender.send(0)
assert await receiver.receive() == 0
Receiving only messages if they are bigger than the previous one. 
from frequenz.channels import Broadcast
from frequenz.channels.experimental import WithPrevious

channel = Broadcast[int](name="example")
receiver = channel.new_receiver().filter(
    WithPrevious(lambda old, new: new > old, first_is_true=False)
)
sender = channel.new_sender()

# This message will skipped as first_is_true is False.
await sender.send(1)

# This message will be received as it is bigger than the previous one (1).
await sender.send(2)
assert await receiver.receive() == 2

# This message will be skipped as it is smaller than the previous one (1).
await sender.send(0)

# This message will be skipped as it is not bigger than the previous one (0).
await sender.send(0)

# This message will be received as it is bigger than the previous one (0).
await sender.send(1)
assert await receiver.receive() == 1

# This message will be received as it is bigger than the previous one (1).
await sender.send(2)
assert await receiver.receive() == 2
Source code in frequenz/channels/experimental/_with_previous.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class WithPrevious(Generic[ChannelMessageT]):
    """A composable predicate to build predicates that can use also the previous message.

    This predicate can be used to filter messages based on a custom condition on the
    previous and current messages. This can be useful in cases where you want to
    process messages only if they satisfy a particular condition with respect to the
    previous message.

    Example: Receiving only messages that are different from the previous one.
        ```python
        from frequenz.channels import Broadcast
        from frequenz.channels.experimental import WithPrevious

        channel = Broadcast[int](name="example")
        receiver = channel.new_receiver().filter(WithPrevious(lambda old, new: old != new))
        sender = channel.new_sender()

        # This message will be received as it is the first message.
        await sender.send(1)
        assert await receiver.receive() == 1

        # This message will be skipped as it equals to the previous one.
        await sender.send(1)

        # This message will be received as it is a different from the previous one.
        await sender.send(0)
        assert await receiver.receive() == 0
        ```

    Example: Receiving only messages if they are bigger than the previous one.
        ```python
        from frequenz.channels import Broadcast
        from frequenz.channels.experimental import WithPrevious

        channel = Broadcast[int](name="example")
        receiver = channel.new_receiver().filter(
            WithPrevious(lambda old, new: new > old, first_is_true=False)
        )
        sender = channel.new_sender()

        # This message will skipped as first_is_true is False.
        await sender.send(1)

        # This message will be received as it is bigger than the previous one (1).
        await sender.send(2)
        assert await receiver.receive() == 2

        # This message will be skipped as it is smaller than the previous one (1).
        await sender.send(0)

        # This message will be skipped as it is not bigger than the previous one (0).
        await sender.send(0)

        # This message will be received as it is bigger than the previous one (0).
        await sender.send(1)
        assert await receiver.receive() == 1

        # This message will be received as it is bigger than the previous one (1).
        await sender.send(2)
        assert await receiver.receive() == 2
        ```
    """

    def __init__(
        self,
        predicate: Callable[[ChannelMessageT, ChannelMessageT], bool],
        /,
        *,
        first_is_true: bool = True,
    ) -> None:
        """Initialize this instance.

        Args:
            predicate: A callable that takes two arguments, the previous message and the
                current message, and returns a boolean indicating whether the current
                message should be received.
            first_is_true: Whether the first message should be considered as satisfying
                the predicate. Defaults to `True`.
        """
        self._predicate = predicate
        self._last_message: ChannelMessageT | _Sentinel = _SENTINEL
        self._first_is_true = first_is_true

    def __call__(self, message: ChannelMessageT) -> bool:
        """Return whether `message` is the first one received or different from the previous one."""

        def is_message(
            value: ChannelMessageT | _Sentinel,
        ) -> TypeGuard[ChannelMessageT]:
            return value is not _SENTINEL

        old_message = self._last_message
        self._last_message = message
        if is_message(old_message):
            return self._predicate(old_message, message)
        return self._first_is_true

    def __str__(self) -> str:
        """Return a string representation of this instance."""
        return f"{type(self).__name__}:{self._predicate.__name__}"

    def __repr__(self) -> str:
        """Return a string representation of this instance."""
        return f"<{type(self).__name__}: {self._predicate!r} first_is_true={self._first_is_true!r}>"
Functions¤
__call__ ¤
__call__(message: ChannelMessageT) -> bool

Return whether message is the first one received or different from the previous one.

Source code in frequenz/channels/experimental/_with_previous.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
def __call__(self, message: ChannelMessageT) -> bool:
    """Return whether `message` is the first one received or different from the previous one."""

    def is_message(
        value: ChannelMessageT | _Sentinel,
    ) -> TypeGuard[ChannelMessageT]:
        return value is not _SENTINEL

    old_message = self._last_message
    self._last_message = message
    if is_message(old_message):
        return self._predicate(old_message, message)
    return self._first_is_true
__init__ ¤
__init__(
    predicate: Callable[
        [ChannelMessageT, ChannelMessageT], bool
    ],
    /,
    *,
    first_is_true: bool = True,
) -> None

Initialize this instance.

PARAMETER DESCRIPTION
predicate

A callable that takes two arguments, the previous message and the current message, and returns a boolean indicating whether the current message should be received.

TYPE: Callable[[ChannelMessageT, ChannelMessageT], bool]

first_is_true

Whether the first message should be considered as satisfying the predicate. Defaults to True.

TYPE: bool DEFAULT: True

Source code in frequenz/channels/experimental/_with_previous.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(
    self,
    predicate: Callable[[ChannelMessageT, ChannelMessageT], bool],
    /,
    *,
    first_is_true: bool = True,
) -> None:
    """Initialize this instance.

    Args:
        predicate: A callable that takes two arguments, the previous message and the
            current message, and returns a boolean indicating whether the current
            message should be received.
        first_is_true: Whether the first message should be considered as satisfying
            the predicate. Defaults to `True`.
    """
    self._predicate = predicate
    self._last_message: ChannelMessageT | _Sentinel = _SENTINEL
    self._first_is_true = first_is_true
__repr__ ¤
__repr__() -> str

Return a string representation of this instance.

Source code in frequenz/channels/experimental/_with_previous.py 
124
125
126
def __repr__(self) -> str:
    """Return a string representation of this instance."""
    return f"<{type(self).__name__}: {self._predicate!r} first_is_true={self._first_is_true!r}>"
__str__ ¤
__str__() -> str

Return a string representation of this instance.

Source code in frequenz/channels/experimental/_with_previous.py 
120
121
122
def __str__(self) -> str:
    """Return a string representation of this instance."""
    return f"{type(self).__name__}:{self._predicate.__name__}"