Make internal variable names private (#213)

The `Anycast` and `Broadcast` implementations were exposing internal
variables to users.  This shouldn't be the case and is fixed in this
PR.

Closes #186

Author: shsms

RELEASE_NOTES.md

@@ -6,7 +6,7 @@ The `Timer` now can be started with a delay.
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+* Internal variable names in the `Anycast` and `Broadcast` implementations are now private.
 
 ## New Features
 

benchmarks/benchmark_broadcast.py

@@ -44,7 +44,7 @@ async def fast_sender(num_messages: int, chan: Sender[int]) -> None:
 
 
 async def benchmark_broadcast(
-    send_msg: Callable[[int, Sender[int]], Coroutine[Any, Any, int]],
+    send_msg: Callable[[int, Sender[int]], Coroutine[Any, Any, None]],
     num_channels: int,
     num_messages: int,
     num_receivers: int,

src/frequenz/channels/_anycast.py

@@ -67,34 +67,34 @@ def __init__(self, maxsize: int = 10) -> None:
         Args:
             maxsize: Size of the channel's buffer.
         """
-        self.limit: int = maxsize
+        self._limit: int = maxsize
         """The maximum number of values that can be stored in the channel's buffer.
 
         If the length of channel's buffer reaches the limit, then the sender
         blocks at the [send()][frequenz.channels.Sender.send] method until
         a value is consumed.
         """
 
-        self.deque: Deque[T] = deque(maxlen=maxsize)
+        self._deque: Deque[T] = deque(maxlen=maxsize)
         """The channel's buffer."""
 
-        self.send_cv: Condition = Condition()
+        self._send_cv: Condition = Condition()
         """The condition to wait for free space in the channel's buffer.
 
         If the channel's buffer is full, then the sender waits for values to
         get consumed using this condition until there's some free space
         available in the channel's buffer.
         """
 
-        self.recv_cv: Condition = Condition()
+        self._recv_cv: Condition = Condition()
         """The condition to wait for values in the channel's buffer.
 
         If the channel's buffer is empty, then the receiver waits for values
         using this condition until there's a value available in the channel's
         buffer.
         """
 
-        self.closed: bool = False
+        self._closed: bool = False
         """Whether the channel is closed."""
 
     async def close(self) -> None:
@@ -109,11 +109,11 @@ async def close(self) -> 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()
+        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.
@@ -164,16 +164,18 @@ async def send(self, msg: T) -> None:
                 A [ChannelClosedError][frequenz.channels.ChannelClosedError] is
                 set as the cause.
         """
-        if self._chan.closed:
+        # pylint: disable=protected-access
+        if self._chan._closed:
             raise SenderError("The channel was closed", self) from ChannelClosedError(
                 self._chan
             )
-        while len(self._chan.deque) == self._chan.deque.maxlen:
-            async with self._chan.send_cv:
-                await self._chan.send_cv.wait()
-        self._chan.deque.append(msg)
-        async with self._chan.recv_cv:
-            self._chan.recv_cv.notify(1)
+        while len(self._chan._deque) == self._chan._deque.maxlen:
+            async with self._chan._send_cv:
+                await self._chan._send_cv.wait()
+        self._chan._deque.append(msg)
+        async with self._chan._recv_cv:
+            self._chan._recv_cv.notify(1)
+        # pylint: enable=protected-access
 
 
 class _Empty:
@@ -213,14 +215,16 @@ async def ready(self) -> bool:
         if self._next is not _Empty:
             return True
 
-        while len(self._chan.deque) == 0:
-            if self._chan.closed:
+        # pylint: disable=protected-access
+        while len(self._chan._deque) == 0:
+            if self._chan._closed:
                 return False
-            async with self._chan.recv_cv:
-                await self._chan.recv_cv.wait()
-        self._next = self._chan.deque.popleft()
-        async with self._chan.send_cv:
-            self._chan.send_cv.notify(1)
+            async with self._chan._recv_cv:
+                await self._chan._recv_cv.wait()
+        self._next = self._chan._deque.popleft()
+        async with self._chan._send_cv:
+            self._chan._send_cv.notify(1)
+        # pylint: enable=protected-access
         return True
 
     def consume(self) -> T:
@@ -233,7 +237,9 @@ def consume(self) -> T:
             ReceiverStoppedError: if the receiver stopped producing messages.
             ReceiverError: if there is some problem with the receiver.
         """
-        if self._next is _Empty and self._chan.closed:
+        if (  # pylint: disable=protected-access
+            self._next is _Empty and self._chan._closed
+        ):
             raise ReceiverStoppedError(self) from ChannelClosedError(self._chan)
 
         assert (

src/frequenz/channels/_broadcast.py

@@ -84,7 +84,7 @@ def __init__(self, name: str, resend_latest: bool = False) -> None:
                 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._name: str = name
         """The name of the broadcast channel.
 
         Only used for debugging purposes.
@@ -93,13 +93,13 @@ def __init__(self, name: str, resend_latest: bool = False) -> None:
         self._resend_latest = resend_latest
         """Whether to resend the latest value to new receivers."""
 
-        self.recv_cv: Condition = Condition()
+        self._recv_cv: Condition = Condition()
         """The condition to wait for data in the channel's buffer."""
 
-        self.receivers: dict[UUID, weakref.ReferenceType[Receiver[T]]] = {}
+        self._receivers: dict[UUID, weakref.ReferenceType[Receiver[T]]] = {}
         """The receivers attached to the channel, indexed by their UUID."""
 
-        self.closed: bool = False
+        self._closed: bool = False
         """Whether the channel is closed."""
 
         self._latest: T | None = None
@@ -117,9 +117,9 @@ async def close(self) -> None:
         immediately.
         """
         self._latest = None
-        self.closed = True
-        async with self.recv_cv:
-            self.recv_cv.notify_all()
+        self._closed = True
+        async with self._recv_cv:
+            self._recv_cv.notify_all()
 
     def new_sender(self) -> Sender[T]:
         """Create a new broadcast sender.
@@ -147,7 +147,7 @@ def new_receiver(self, name: str | None = None, maxsize: int = 50) -> Receiver[T
         if name is None:
             name = str(uuid)
         recv: Receiver[T] = Receiver(uuid, name, maxsize, self)
-        self.receivers[uuid] = weakref.ref(recv)
+        self._receivers[uuid] = weakref.ref(recv)
         if self._resend_latest and self._latest is not None:
             recv.enqueue(self._latest)
         return recv
@@ -193,23 +193,24 @@ async def send(self, msg: T) -> None:
                 A [ChannelClosedError][frequenz.channels.ChannelClosedError] is
                 set as the cause.
         """
-        if self._chan.closed:
+        # pylint: disable=protected-access
+        if self._chan._closed:
             raise SenderError("The channel was closed", self) from ChannelClosedError(
                 self._chan
             )
-        # pylint: disable=protected-access
         self._chan._latest = msg
         stale_refs = []
-        for name, recv_ref in self._chan.receivers.items():
+        for name, recv_ref in self._chan._receivers.items():
             recv = recv_ref()
             if recv is None:
                 stale_refs.append(name)
                 continue
             recv.enqueue(msg)
         for name in stale_refs:
-            del self._chan.receivers[name]
-        async with self._chan.recv_cv:
-            self._chan.recv_cv.notify_all()
+            del self._chan._receivers[name]
+        async with self._chan._recv_cv:
+            self._chan._recv_cv.notify_all()
+        # pylint: enable=protected-access
 
 
 class Receiver(BaseReceiver[T]):
@@ -271,7 +272,7 @@ def enqueue(self, msg: T) -> None:
             self._q.popleft()
             logger.warning(
                 "Broadcast receiver [%s:%s] is full. Oldest message was dropped.",
-                self._chan.name,
+                self._chan._name,  # pylint: disable=protected-access
                 self._name,
             )
         self._q.append(msg)
@@ -307,18 +308,22 @@ async def ready(self) -> bool:
         #
         # The condition also makes sure that if there are already messages ready to be
         # consumed, then we return immediately.
+        # pylint: disable=protected-access
         while len(self._q) == 0:
-            if self._chan.closed:
+            if self._chan._closed:
                 return False
-            async with self._chan.recv_cv:
-                await self._chan.recv_cv.wait()
+            async with self._chan._recv_cv:
+                await self._chan._recv_cv.wait()
         return True
+        # pylint: enable=protected-access
 
     def _deactivate(self) -> None:
         """Set the receiver as inactive and remove it from the channel."""
         self._active = False
-        if self._uuid in self._chan.receivers:
-            del self._chan.receivers[self._uuid]
+        # pylint: disable=protected-access
+        if self._uuid in self._chan._receivers:
+            del self._chan._receivers[self._uuid]
+        # pylint: enable=protected-access
 
     def consume(self) -> T:
         """Return the latest value once `ready` is complete.
@@ -337,7 +342,7 @@ def consume(self) -> T:
                 self,
             )
 
-        if not self._q and self._chan.closed:
+        if not self._q and self._chan._closed:  # pylint: disable=protected-access
             raise ReceiverStoppedError(self) from ChannelClosedError(self._chan)
 
         assert self._q, "`consume()` must be preceded by a call to `ready()`"

tests/test_broadcast.py

@@ -265,10 +265,12 @@ async def test_broadcast_receiver_drop() -> None:
     assert 10 == await receiver1.receive()
     assert 10 == await receiver2.receive()
 
-    assert len(chan.receivers) == 2
+    # pylint: disable=protected-access
+    assert len(chan._receivers) == 2
 
     del receiver2
 
     await sender.send(20)
 
-    assert len(chan.receivers) == 1
+    assert len(chan._receivers) == 1
+    # pylint: enable=protected-access