Rename Receiver methods _ready, _get to ready, consume

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/channels/anycast.py

@@ -164,7 +164,7 @@ def __init__(self, chan: Anycast[T]) -> None:
         self._chan = chan
         self._next: Optional[T] = None
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 
         Raises:
@@ -183,17 +183,19 @@ async def _ready(self) -> None:
         async with self._chan.send_cv:
             self._chan.send_cv.notify(1)
 
-    def _get(self) -> T:
-        """Return the latest value once `_ready()` is complete.
+    def consume(self) -> T:
+        """Return the latest value once `ready()` is complete.
 
         Raises:
-            EOFError: When called before a call to `_ready()` finishes.
+            EOFError: When called before a call to `ready()` finishes.
 
         Returns:
             The next value that was received.
         """
         if self._next is None:
-            raise EOFError("_get was called before a call to _ready finished.")
+            raise EOFError(
+                "`consume()` was called before a call to `ready()` finished."
+            )
         next_val = self._next
         self._next = None
         return next_val

src/frequenz/channels/base_classes.py

@@ -69,25 +69,25 @@ async def __anext__(self) -> T:
         Raises:
             StopAsyncIteration: if the underlying channel is closed.
         """
-        await self._ready()
-        return self._get()
+        await self.ready()
+        return self.consume()
 
     @abstractmethod
-    async def _ready(self) -> None:
+    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
-        `_get()`.
+        Once a call to `ready()` has finished, the value should be read with a call to
+        `consume()`.
 
         Raises:
             StopAsyncIteration: if the underlying channel is closed.
         """
 
     @abstractmethod
-    def _get(self) -> T:
-        """Return the latest value once `_ready` is complete.
+    def consume(self) -> T:
+        """Return the latest value once `ready()` is complete.
 
-        `_ready()` must be called before each call to `_get()`.
+        `ready()` must be called before each call to `consume()`.
 
         Returns:
             The next value received.
@@ -193,14 +193,14 @@ def __init__(self, recv: Receiver[T], transform: Callable[[T], U]) -> None:
         self._recv = recv
         self._transform = transform
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait until the receiver is ready with a value."""
-        await self._recv._ready()  # pylint: disable=protected-access
+        await self._recv.ready()  # pylint: disable=protected-access
 
-    def _get(self) -> U:
-        """Return a transformed value once `_ready()` is complete.
+    def consume(self) -> U:
+        """Return a transformed value once `ready()` is complete.
 
         Returns:
             The next value that was received.
         """
-        return self._transform(self._recv._get())  # pylint: disable=protected-access
+        return self._transform(self._recv.consume())  # pylint: disable=protected-access

src/frequenz/channels/bidirectional.py

@@ -82,14 +82,14 @@ async def send(self, msg: T) -> bool:
         """
         return await self._sender.send(msg)
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait until the receiver is ready with a value."""
-        await self._receiver._ready()  # pylint: disable=protected-access
+        await self._receiver.ready()  # pylint: disable=protected-access
 
-    def _get(self) -> U:
+    def consume(self) -> U:
         """Return the latest value once `_ready` is complete.
 
         Returns:
             The next value that was received.
         """
-        return self._receiver._get()  # pylint: disable=protected-access
+        return self._receiver.consume()  # pylint: disable=protected-access

src/frequenz/channels/broadcast.py

@@ -249,11 +249,11 @@ def __len__(self) -> int:
         """
         return len(self._q)
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 
         Raises:
-            EOFError: When called before a call to `_ready()` finishes.
+            EOFError: When called before a call to `ready()` finishes.
             StopAsyncIteration: if the underlying channel is closed.
         """
         if not self._active:
@@ -269,8 +269,8 @@ async def _ready(self) -> None:
             async with self._chan.recv_cv:
                 await self._chan.recv_cv.wait()
 
-    def _get(self) -> T:
-        """Return the latest value once `_ready` is complete.
+    def consume(self) -> T:
+        """Return the latest value once `ready` is complete.
 
         Returns:
             The next value that was received.

src/frequenz/channels/merge.py

@@ -44,7 +44,7 @@ def __del__(self) -> None:
         for task in self._pending:
             task.cancel()
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 
         Raises:
@@ -74,16 +74,16 @@ async def _ready(self) -> None:
                     asyncio.create_task(self._receivers[name].__anext__(), name=name)
                 )
 
-    def _get(self) -> T:
-        """Return the latest value once `_ready` is complete.
+    def consume(self) -> T:
+        """Return the latest value once `ready` is complete.
 
         Raises:
-            EOFError: When called before a call to `_ready()` finishes.
+            EOFError: When called before a call to `ready()` finishes.
 
         Returns:
             The next value that was received.
         """
         if not self._results:
-            raise EOFError("_get called before _ready finished.")
+            raise EOFError("`consume` called before `ready` finished.")
 
         return self._results.popleft()

src/frequenz/channels/merge_named.py

@@ -31,7 +31,7 @@ def __del__(self) -> None:
         for task in self._pending:
             task.cancel()
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait until there's a message in any of the channels.
 
         Raises:
@@ -61,11 +61,11 @@ async def _ready(self) -> None:
                     asyncio.create_task(self._receivers[name].__anext__(), name=name)
                 )
 
-    def _get(self) -> Tuple[str, T]:
-        """Return the latest value once `_ready` is complete.
+    def consume(self) -> Tuple[str, T]:
+        """Return the latest value once `ready` is complete.
 
         Raises:
-            EOFError: When called before a call to `_ready()` finishes.
+            EOFError: When called before a call to `ready()` finishes.
 
         Returns:
             The next value that was received, along with its name.

src/frequenz/channels/select.py

@@ -50,7 +50,7 @@ def get(self) -> _Selected:
         """
         if self.recv is None:
             return _Selected(None)
-        return _Selected(self.recv._get())  # pylint: disable=protected-access
+        return _Selected(self.recv.consume())  # pylint: disable=protected-access
 
 
 class Select:
@@ -93,9 +93,7 @@ def __init__(self, **kwargs: Receiver[Any]) -> None:
         self._pending: Set[asyncio.Task[None]] = set()
 
         for name, recv in self._receivers.items():
-            # can replace __anext__() to anext() (Only Python 3.10>=)
-            ready = recv._ready()  # pylint: disable=unnecessary-dunder-call
-            self._pending.add(asyncio.create_task(ready, name=name))
+            self._pending.add(asyncio.create_task(recv.ready(), name=name))
 
         self._ready_count = 0
         self._prev_ready_count = 0
@@ -124,7 +122,7 @@ async def ready(self) -> bool:
                     if value is not None:
                         dropped_names.append(name)
                         if value.recv is not None:
-                            value.recv._get()  # pylint: disable=protected-access
+                            value.recv.consume()
                         self._result[name] = None
                 self._ready_count = 0
                 self._prev_ready_count = 0
@@ -159,7 +157,7 @@ async def ready(self) -> bool:
             # don't add a task for it again.
             if result is None:
                 continue
-            ready = recv._ready()  # pylint: disable=protected-access
+            ready = recv.ready()
             self._pending.add(asyncio.create_task(ready, name=name))
         return True
 

src/frequenz/channels/utils/file_watcher.py

@@ -64,7 +64,7 @@ def __del__(self) -> None:
         """
         self._stop_event.set()
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Wait for the next file event and return its path.
 
         Raises:
@@ -79,7 +79,15 @@ async def _ready(self) -> None:
 
         self._changes = await self._awatch.__anext__()
 
-    def _get(self) -> pathlib.Path:
+    def consume(self) -> pathlib.Path:
+        """Return the latest change once `ready` is complete.
+
+        Raises:
+            StopAsyncIteration: When the channel is closed.
+
+        Returns:
+            The next change that was received.
+        """
         change = self._changes.pop()
         # Tuple of (Change, path) returned by watchfiles
         if change is None or len(change) != 2:

src/frequenz/channels/utils/timer.py

@@ -77,7 +77,7 @@ def stop(self) -> None:
         """
         self._stopped = True
 
-    async def _ready(self) -> None:
+    async def ready(self) -> None:
         """Return the current time (in UTC) once the next tick is due.
 
         Raises:
@@ -88,10 +88,6 @@ async def _ready(self) -> None:
             The time of the next tick in UTC or `None` if
                 [stop()][frequenz.channels.Timer.stop] has been called on the
                 timer.
-
-        Changelog:
-            * **v0.11.0:** Returns a timezone-aware datetime with UTC timezone
-              instead of a native datetime object.
         """
         # if there are messages waiting to be consumed, return immediately.
         if self._now is not None:
@@ -109,17 +105,21 @@ async def _ready(self) -> None:
 
         self._next_msg_time = self._now + self._interval
 
-    def _get(self) -> datetime:
-        """Return the latest value once `_ready` is complete.
+    def consume(self) -> datetime:
+        """Return the latest value once `ready` is complete.
 
         Raises:
-            EOFError: When called before a call to `_ready()` finishes.
+            EOFError: When called before a call to `ready()` finishes.
 
         Returns:
             The timestamp for the next tick.
+
+        Changelog:
+            * **v0.11.0:** Returns a timezone-aware datetime with UTC timezone
+              instead of a native datetime object.
         """
         if self._now is None:
-            raise EOFError("_get called before _ready finished")
+            raise EOFError("`consume` called before `ready` finished")
         now = self._now
         self._now = None
         return now