improved grace period handling

Author: daniel-sanche

google/cloud/bigtable/data/_async/client.py

@@ -480,12 +480,8 @@ async def _manage_channel(
             old_channel = super_channel.swap_channel(new_channel)
             self._invalidate_channel_stubs()
             # give old_channel a chance to complete existing rpcs
-            if CrossSync.is_async:
-                await old_channel.close(grace_period)
-            else:
-                if grace_period:
-                    self._is_closed.wait(grace_period)  # type: ignore
-                old_channel.close()  # type: ignore
+            await CrossSync.event_wait(self._is_closed, grace_period, async_break_early=False)
+            await old_channel.close()
             # subtract the time spent waiting for the channel to be replaced
             next_refresh = random.uniform(refresh_interval_min, refresh_interval_max)
             next_sleep = max(next_refresh - (time.monotonic() - start_timestamp), 0)

google/cloud/bigtable/data/_sync_autogen/client.py

@@ -363,8 +363,9 @@ def _manage_channel(
             self._ping_and_warm_instances(channel=new_channel)
             old_channel = super_channel.swap_channel(new_channel)
             self._invalidate_channel_stubs()
-            if grace_period:
-                self._is_closed.wait(grace_period)
+            CrossSync._Sync_Impl.event_wait(
+                self._is_closed, grace_period, async_break_early=False
+            )
             old_channel.close()
             next_refresh = random.uniform(refresh_interval_min, refresh_interval_max)
             next_sleep = max(next_refresh - (time.monotonic() - start_timestamp), 0)