Switch callbacks to coroutines

Instead of passing Callable[...,Coroutine] functions as callbacks, pass
Coroutines directly. This allows the caller to supply arbitrary context
associated with the Coroutine, such as the ApiClient used to establish
the leader election.

Author: JacobHenner

examples/leaderelection.py

@@ -62,8 +62,8 @@ async def example_end_func():
             lease_duration=17,
             renew_deadline=15,
             retry_period=5,
-            onstarted_leading=example_start_func,
-            onstopped_leading=example_end_func,
+            onstarted_leading=example_start_func(),
+            onstopped_leading=example_end_func(),
         )
 
         # Enter leader election

kubernetes_asyncio/leaderelection/electionconfig.py

@@ -13,20 +13,25 @@
 # limitations under the License.
 
 import logging
+from collections.abc import Coroutine  # noqa:F401
 
 logging.basicConfig(level=logging.INFO)
 
 
 class Config:
     # Validate config, exit if an error is detected
+
+    # onstarted_leading and onstopped_leading are defined as coroutines rather
+    # than callables in order to faciliate passing context. For example, this
+    # allows the ApiClient used by the leader election to be shared and reused.
     def __init__(
         self,
         lock,
         lease_duration,
         renew_deadline,
         retry_period,
-        onstarted_leading,
-        onstopped_leading,
+        onstarted_leading,  # type: Coroutine
+        onstopped_leading=None,  # type: Coroutine | None
     ):
         self.jitter_factor = 1.2
 
@@ -59,11 +64,4 @@ def __init__(
             raise ValueError("callback onstarted_leading cannot be None")
         self.onstarted_leading = onstarted_leading
 
-        if onstopped_leading is None:
-            self.onstopped_leading = self.on_stoppedleading_callback
-        else:
-            self.onstopped_leading = onstopped_leading
-
-    # Default callback for when the current candidate if a leader, stops leading
-    def on_stoppedleading_callback(self):
-        logging.info("{} stopped leading".format(self.lock.identity))
+        self.onstopped_leading = onstopped_leading

kubernetes_asyncio/leaderelection/leaderelection.py

@@ -59,7 +59,7 @@ async def run(self):
                 )
             )
 
-            task = asyncio.create_task(self.election_config.onstarted_leading())
+            task = asyncio.create_task(self.election_config.onstarted_leading)
 
             await self.renew_loop()
 
@@ -71,7 +71,8 @@ async def run(self):
             # Failed to update lease, run onstopped_leading callback. This is
             # preserved in order to continue to provide an interface similar to
             # the one provided by `kubernetes-client/python`.
-            await self.election_config.onstopped_leading()
+            if self.election_config.onstopped_leading is not None:
+                await self.election_config.onstopped_leading
 
     async def acquire(self):
         # Follower

kubernetes_asyncio/leaderelection/leaderelection_test.py

@@ -62,8 +62,8 @@ async def on_stopped_leading():
             lease_duration=2.5,
             renew_deadline=2,
             retry_period=1.5,
-            onstarted_leading=on_started_leading,
-            onstopped_leading=on_stopped_leading,
+            onstarted_leading=on_started_leading(),
+            onstopped_leading=on_stopped_leading(),
         )
 
         # Enter leader election
@@ -116,8 +116,8 @@ async def on_stopped_leading_A():
             lease_duration=2.5,
             renew_deadline=2,
             retry_period=1.5,
-            onstarted_leading=on_started_leading_A,
-            onstopped_leading=on_stopped_leading_A,
+            onstarted_leading=on_started_leading_A(),
+            onstopped_leading=on_stopped_leading_A(),
         )
 
         def on_create_B():
@@ -153,8 +153,8 @@ async def on_stopped_leading_B():
             lease_duration=2.5,
             renew_deadline=2,
             retry_period=1.5,
-            onstarted_leading=on_started_leading_B,
-            onstopped_leading=on_stopped_leading_B,
+            onstarted_leading=on_started_leading_B(),
+            onstopped_leading=on_stopped_leading_B(),
         )
 
         mock_lock_B.leader_record = mock_lock_A.leader_record
@@ -246,8 +246,8 @@ async def on_stopped_leading():
             lease_duration=2.5,
             renew_deadline=2,
             retry_period=1.5,
-            onstarted_leading=on_started_leading,
-            onstopped_leading=on_stopped_leading,
+            onstarted_leading=on_started_leading(),
+            onstopped_leading=on_stopped_leading(),
         )
 
         # Enter leader election