Make Receiver.close method async

This makes it consistent with the channel close method, which is
async.  It also makes it easy to fully stop any pending tasks, as
is the case with `merge`.

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

Author: shsms

src/frequenz/channels/_anycast.py

@@ -456,7 +456,7 @@ def consume(self) -> _T:
         return next_val
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close this receiver.
 
         After closing, the receiver will not be able to receive any more messages.

src/frequenz/channels/_broadcast.py

@@ -496,7 +496,7 @@ def consume(self) -> _T:
         return self._q.popleft()
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close the receiver.
 
         After calling this method, new messages will not be received.  Once the

src/frequenz/channels/_merge.py

@@ -192,7 +192,7 @@ def consume(self) -> ReceiverMessageT_co:
         return self._results.popleft()
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close the receiver.
 
         After calling this method, new messages will not be received.  Once the
@@ -202,9 +202,13 @@ def close(self) -> None:
         for task in self._pending:
             if not task.done() and task.get_loop().is_running():
                 task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
         self._pending = set()
         for recv in self._receivers.values():
-            recv.close()
+            await recv.close()
 
     def __str__(self) -> str:
         """Return a string representation of this receiver."""

src/frequenz/channels/_receiver.py

@@ -218,7 +218,7 @@ def consume(self) -> ReceiverMessageT_co:
         """
 
     @abstractmethod
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close the receiver.
 
         After calling this method, new messages will not be received.  Once the
@@ -474,14 +474,14 @@ def consume(self) -> MappedMessageT_co:  # noqa: DOC502
         return self._mapping_function(self._receiver.consume())
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close the receiver.
 
         After calling this method, new messages will not be received.  Once the
         receiver's buffer is drained, trying to receive a message will raise a
         [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
         """
-        self._receiver.close()
+        await self._receiver.close()
 
     def __str__(self) -> str:
         """Return a string representation of the mapper."""
@@ -573,14 +573,14 @@ def consume(self) -> ReceiverMessageT_co:
         return message
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close the receiver.
 
         After calling this method, new messages will not be received.  Once the
         receiver's buffer is drained, trying to receive a message will raise a
         [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
         """
-        self._receiver.close()
+        await self._receiver.close()
 
     def __str__(self) -> str:
         """Return a string representation of the filter."""

src/frequenz/channels/event.py

@@ -173,7 +173,7 @@ def consume(self) -> None:
         self._event.clear()
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close this receiver."""
         self.stop()
 

src/frequenz/channels/file_watcher.py

@@ -233,7 +233,7 @@ def consume(self) -> Event:
         return Event(type=EventType(change), path=pathlib.Path(path_str))
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close this receiver."""
         self._stop_event.set()
 

src/frequenz/channels/timer.py

@@ -746,7 +746,7 @@ def consume(self) -> timedelta:
         return drift
 
     @override
-    def close(self) -> None:
+    async def close(self) -> None:
         """Close the timer."""
         self.stop()