Add stop method to MetricFetcher

Stop fallback metric fetcher in the `stop` method.

Author: ela-kotulska-frequenz

src/frequenz/sdk/timeseries/formula_engine/_formula_engine.py

@@ -117,9 +117,10 @@ async def stop(self) -> None:
         if self._task is None:
             return
         await cancel_and_await(self._task)
+
         _, fetchers = self._builder.finalize()
         for fetcher in fetchers.values():
-            fetcher.stream.close()
+            await fetcher.stop()
 
     @classmethod
     def from_receiver(

src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_fallback_formula_metric_fetcher.py

@@ -77,3 +77,14 @@ def consume(self) -> Sample[QuantityT]:
         ), f"Fallback metric fetcher: {self.name} was not started"
 
         return self._receiver.consume()
+
+    async def stop(self) -> None:
+        if self._formula_engine is None:
+            return
+        await self._formula_engine.stop()
+        self._formula_engine = None
+
+        if self._receiver is None:
+            return
+        self._receiver.close()
+        self._receiver = None

src/frequenz/sdk/timeseries/formula_engine/_formula_steps.py

@@ -368,6 +368,10 @@ def is_running(self) -> bool:
     def start(self) -> None:
         """Initialize the metric fetcher and start fetching samples."""
 
+    @abstractmethod
+    async def stop(self) -> None:
+        """Stope the fetcher if is running."""
+
 
 class MetricFetcher(Generic[QuantityT], FormulaStep):
     """A formula step for fetching a value from a metric Receiver."""
@@ -406,6 +410,11 @@ def stream(self) -> Receiver[Sample[QuantityT]]:
         """
         return self._stream
 
+    async def stop(self) -> None:
+        self.stream.close()
+        if self._fallback:
+            await self._fallback.stop()
+
     def stream_name(self) -> str:
         """Return the name of the stream.