Port all existing uses of Selector to select

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

Author: llucax

.github/ISSUE_TEMPLATE/bug.yml

@@ -49,7 +49,7 @@ body:
         - Unit, integration and performance tests (part:tests)
         - Build script, CI, dependencies, etc. (part:tooling)
         - Channels, `Broadcast`, `Bidirectional`, etc. (part:channels)
-        - Selector (part:selector)
+        - Select (part:select)
         - Utility receivers, `Merge`, etc. (part:receivers)
     validations:
       required: true

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 a [Selector][frequenz.channels.util.Selector],
+    simultaneously using [select][frequenz.channels.util.select],
     [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 a [Selector][frequenz.channels.util.Selector],
+    simultaneously using [select][frequenz.channels.util.select],
     [Merge][frequenz.channels.util.Merge] or
     [MergeNamed][frequenz.channels.util.MergeNamed].
 

src/frequenz/channels/util/_timer.py

@@ -307,12 +307,12 @@ class Timer(Receiver[timedelta]):
             print(f"The timer has triggered {drift=}")
         ```
 
-        But you can also use a [`Selector`][frequenz.channels.util.Selector] to combine
+        But you can also use a [`select`][frequenz.channels.util.select] to combine
         it with other receivers, and even start it (semi) manually:
 
         ```python
         import logging
-        from frequenz.channels.util import Selector, selected_from
+        from frequenz.channels.util import select, selected_from
         from frequenz.channels import Broadcast
 
         timer = Timer.timeout(timedelta(seconds=1.0), auto_start=False)
@@ -322,22 +322,21 @@ class Timer(Receiver[timedelta]):
         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()`).
-        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}%")
+        async for selected in select(battery_data, timer):
+            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 Selector, selected_from
+        from frequenz.channels.util import select, selected_from
         from frequenz.channels import Broadcast
 
         def process_data(data: int):
@@ -351,21 +350,20 @@ def do_heavy_processing(data: int):
         chan2 = Broadcast[int]("input-chan-2")
         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()
-                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")
+        async for selected in select(battery_data, heavy_process, timer):
+            if selected_from(selected, battery_data):
+                if selected.was_closed():
+                    logging.warning("battery channel closed")
+                    continue
+                process_data(selected.value)
+                timer.reset()
+            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, Selector, Timer, selected_from
+from frequenz.channels.util import FileWatcher, Timer, select, selected_from
 
 
 @pytest.mark.integration
@@ -27,23 +27,20 @@ async def test_file_watcher(tmp_path: pathlib.Path) -> None:
     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
+    async for selected in select(file_watcher, timer):
+        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
 
@@ -85,23 +82,22 @@ async def test_file_watcher_deletes(tmp_path: pathlib.Path) -> None:
     # W: Write
     # D: Delete
     # E: FileWatcher Event
-    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
+    async for selected in select(file_watcher, write_timer, deletion_timer):
+        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