Update existing code to use Selector instead of Select

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/channels/_anycast.py

@@ -28,7 +28,7 @@ class Anycast(Generic[T]):
     thread-safe.
 
     When there are multiple channel receivers, they can be awaited
-    simultaneously using [Select][frequenz.channels.util.Select],
+    simultaneously using a [Selector][frequenz.channels.util.Selector],
     [Merge][frequenz.channels.util.Merge] or
     [MergeNamed][frequenz.channels.util.MergeNamed].
 

src/frequenz/channels/_broadcast.py

@@ -38,7 +38,7 @@ class Broadcast(Generic[T]):
     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],
+    simultaneously using a [Selector][frequenz.channels.util.Selector],
     [Merge][frequenz.channels.util.Merge] or
     [MergeNamed][frequenz.channels.util.MergeNamed].
 

src/frequenz/channels/util/_timer.py

@@ -307,37 +307,37 @@ class Timer(Receiver[timedelta]):
             print(f"The timer has triggered {drift=}")
         ```
 
-        But you can also use [`Select`][frequenz.channels.util.Select] to combine it
-        with other receivers, and even start it (semi) manually:
+        But you can also use a [`Selector`][frequenz.channels.util.Selector] to combine
+        it with other receivers, and even start it (semi) manually:
 
         ```python
         import logging
-        from frequenz.channels.util import Select
+        from frequenz.channels.util import Selector, selected_from
         from frequenz.channels import Broadcast
 
         timer = Timer.timeout(timedelta(seconds=1.0), auto_start=False)
         chan = Broadcast[int]("input-chan")
-        receiver1 = chan.new_receiver()
+        battery_data = chan.new_receiver()
 
         timer = Timer.timeout(timedelta(seconds=1.0), auto_start=False)
         # Do some other initialization, the timer will start automatically if
         # a message is awaited (or manually via `reset()`).
-        select = Select(bat_1=receiver1, timer=timer)
-        while await select.ready():
-            if msg := select.bat_1:
-                if val := msg.inner:
-                    battery_soc = val
-                else:
-                    logging.warning("battery channel closed")
-            elif drift := select.timer:
-                # Print some regular battery data
-                print(f"Battery is charged at {battery_soc}%")
+        async with Selector(battery_data, timer) as selector:
+            async for selected in selector:
+                if selected_from(selected, battery_data):
+                    if selected.was_closed():
+                        logging.warning("battery channel closed")
+                        continue
+                    battery_soc = selected.value
+                elif selected_from(selected, timer):
+                    # Print some regular battery data
+                    print(f"Battery is charged at {battery_soc}%")
         ```
 
     Example: Timeout example
         ```python
         import logging
-        from frequenz.channels.util import Select
+        from frequenz.channels.util import Selector, selected_from
         from frequenz.channels import Broadcast
 
         def process_data(data: int):
@@ -349,23 +349,23 @@ def do_heavy_processing(data: int):
         timer = Timer.timeout(timedelta(seconds=1.0), auto_start=False)
         chan1 = Broadcast[int]("input-chan-1")
         chan2 = Broadcast[int]("input-chan-2")
-        receiver1 = chan1.new_receiver()
-        receiver2 = chan2.new_receiver()
-        select = Select(bat_1=receiver1, heavy_process=receiver2, timeout=timer)
-        while await select.ready():
-            if msg := select.bat_1:
-                if val := msg.inner:
-                    process_data(val)
+        battery_data = chan1.new_receiver()
+        heavy_process = chan2.new_receiver()
+        async with Selector(battery_data, heavy_process, timer) as selector:
+            async for selected in selector:
+                if selected_from(selected, battery_data):
+                    if selected.was_closed():
+                        logging.warning("battery channel closed")
+                        continue
+                    process_data(selected.value)
                     timer.reset()
-                else:
-                    logging.warning("battery channel closed")
-            if msg := select.heavy_process:
-                if val := msg.inner:
-                    do_heavy_processing(val)
-                else:
-                    logging.warning("processing channel closed")
-            elif drift := select.timeout:
-                logging.warning("No data received in time")
+                elif selected_from(selected, heavy_process):
+                    if selected.was_closed():
+                        logging.warning("processing channel closed")
+                        continue
+                    do_heavy_processing(selected.value)
+                elif selected_from(selected, timer):
+                    logging.warning("No data received in time")
         ```
 
         In this case `do_heavy_processing` might take 2 seconds, and we don't

tests/utils/test_integration.py

@@ -9,7 +9,7 @@
 
 import pytest
 
-from frequenz.channels.util import FileWatcher, Select, Timer
+from frequenz.channels.util import FileWatcher, Selector, Timer, selected_from
 
 
 @pytest.mark.integration
@@ -20,29 +20,30 @@ async def test_file_watcher(tmp_path: pathlib.Path) -> None:
         tmp_path: A tmp directory to run the file watcher on. Created by pytest.
     """
     filename = tmp_path / "test-file"
-    file_watcher = FileWatcher(paths=[str(tmp_path)])
 
     number_of_writes = 0
     expected_number_of_writes = 3
 
-    select = Select(
-        timer=Timer.timeout(timedelta(seconds=0.1)),
-        file_watcher=file_watcher,
-    )
-    while await select.ready():
-        if msg := select.timer:
-            filename.write_text(f"{msg.inner}")
-        elif msg := select.file_watcher:
-            event_type = (
-                FileWatcher.EventType.CREATE
-                if number_of_writes == 0
-                else FileWatcher.EventType.MODIFY
-            )
-            assert msg.inner == FileWatcher.Event(type=event_type, path=filename)
-            number_of_writes += 1
-            # After receiving a write 3 times, unsubscribe from the writes channel
-            if number_of_writes == expected_number_of_writes:
-                break
+    file_watcher = FileWatcher(paths=[str(tmp_path)])
+    timer = Timer.timeout(timedelta(seconds=0.1))
+
+    async with Selector(file_watcher, timer) as selector:
+        async for selected in selector:
+            if selected_from(selected, timer):
+                filename.write_text(f"{selected.value}")
+            elif selected_from(selected, file_watcher):
+                event_type = (
+                    FileWatcher.EventType.CREATE
+                    if number_of_writes == 0
+                    else FileWatcher.EventType.MODIFY
+                )
+                assert selected.value == FileWatcher.Event(
+                    type=event_type, path=filename
+                )
+                number_of_writes += 1
+                # After receiving a write 3 times, unsubscribe from the writes channel
+                if number_of_writes == expected_number_of_writes:
+                    break
 
     assert number_of_writes == expected_number_of_writes
 
@@ -61,12 +62,9 @@ async def test_file_watcher_deletes(tmp_path: pathlib.Path) -> None:
     file_watcher = FileWatcher(
         paths=[str(tmp_path)], event_types={FileWatcher.EventType.DELETE}
     )
+    write_timer = Timer.timeout(timedelta(seconds=0.1))
+    deletion_timer = Timer.timeout(timedelta(seconds=0.25))
 
-    select = Select(
-        write_timer=Timer.timeout(timedelta(seconds=0.1)),
-        deletion_timer=Timer.timeout(timedelta(seconds=0.25)),
-        watcher=file_watcher,
-    )
     number_of_write = 0
     number_of_deletes = 0
     number_of_events = 0
@@ -87,22 +85,23 @@ async def test_file_watcher_deletes(tmp_path: pathlib.Path) -> None:
     # W: Write
     # D: Delete
     # E: FileWatcher Event
-    while await select.ready():
-        if msg := select.write_timer:
-            if number_of_write >= 2 and number_of_events == 0:
-                continue
-            filename.write_text(f"{msg.inner}")
-            number_of_write += 1
-        elif _ := select.deletion_timer:
-            # Avoid removing the file twice
-            if not pathlib.Path(filename).is_file():
-                continue
-            os.remove(filename)
-            number_of_deletes += 1
-        elif _ := select.watcher:
-            number_of_events += 1
-            if number_of_events >= 2:
-                break
+    async with Selector(file_watcher, write_timer, deletion_timer) as selector:
+        async for selected in selector:
+            if selected_from(selected, write_timer):
+                if number_of_write >= 2 and number_of_events == 0:
+                    continue
+                filename.write_text(f"{selected.value}")
+                number_of_write += 1
+            elif selected_from(selected, deletion_timer):
+                # Avoid removing the file twice
+                if not pathlib.Path(filename).is_file():
+                    continue
+                os.remove(filename)
+                number_of_deletes += 1
+            elif selected_from(selected, file_watcher):
+                number_of_events += 1
+                if number_of_events >= 2:
+                    break
 
     assert number_of_deletes == 2
     # Can be more because the watcher could take some time to trigger