Merge pull request #21885 from lidizheng/aio-channel-ready

[Aio] Add channel_ready helper function

src/python/grpcio/grpc/experimental/aio/_channel.py

@@ -461,6 +461,13 @@ class Channel:
         assert await self._channel.watch_connectivity_state(
             last_observed_state.value[0], None)
 
+    async def channel_ready(self) -> None:
+        """Creates a coroutine that ends when a Channel is ready."""
+        state = self.get_state(try_to_connect=True)
+        while state != grpc.ChannelConnectivity.READY:
+            await self.wait_for_state_change(state)
+            state = self.get_state(try_to_connect=True)
+
     def unary_unary(
             self,
             method: Text,

src/python/grpcio_tests/tests_aio/tests.json

@@ -9,6 +9,7 @@
   "unit.call_test.TestUnaryStreamCall",
   "unit.call_test.TestUnaryUnaryCall",
   "unit.channel_argument_test.TestChannelArgument",
+  "unit.channel_ready_test.TestChannelReady",
   "unit.channel_test.TestChannel",
   "unit.close_channel_test.TestCloseChannel",
   "unit.close_channel_test.TestOngoingCalls",

src/python/grpcio_tests/tests_aio/unit/channel_ready_test.py

@@ -0,0 +1,67 @@
+# Copyright 2020 The gRPC Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Testing the channel_ready function."""
+
+import asyncio
+import gc
+import logging
+import time
+import unittest
+
+import grpc
+from grpc.experimental import aio
+
+from tests.unit.framework.common import get_socket, test_constants
+from tests_aio.unit import _common
+from tests_aio.unit._test_base import AioTestBase
+from tests_aio.unit._test_server import start_test_server
+
+
+class TestChannelReady(AioTestBase):
+
+    async def setUp(self):
+        address, self._port, self._socket = get_socket(listen=False)
+        self._channel = aio.insecure_channel(f"{address}:{self._port}")
+        self._socket.close()
+
+    async def tearDown(self):
+        await self._channel.close()
+
+    async def test_channel_ready_success(self):
+        # Start `channel_ready` as another Task
+        channel_ready_task = self.loop.create_task(
+            self._channel.channel_ready())
+
+        # Wait for TRANSIENT_FAILURE
+        await _common.block_until_certain_state(
+            self._channel, grpc.ChannelConnectivity.TRANSIENT_FAILURE)
+
+        try:
+            # Start the server
+            _, server = await start_test_server(port=self._port)
+
+            # The RPC should recover itself
+            await channel_ready_task
+        finally:
+            await server.stop(None)
+
+    async def test_channel_ready_blocked(self):
+        with self.assertRaises(asyncio.TimeoutError):
+            await asyncio.wait_for(self._channel.channel_ready(),
+                                   test_constants.SHORT_TIMEOUT)
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    unittest.main(verbosity=2)