class_docstr: Async io subscriber to GCS.
  
  Usage example common to Aio subscribers:
      subscriber = GcsAioXxxSubscriber(address="...")
      await subscriber.subscribe()
      while running:
          ...... = await subscriber.poll()
          ......
      await subscriber.close()
class_docstr_tok: ["Async","io","subscriber","to","GCS",".","Usage","example","common","to","Aio","subscribers",":","subscriber","=","GcsAioXxxSubscriber","(","address=","''","...","''",")","await","subscriber.subscribe","(",")","while","running",":","......","=","await","subscriber.poll","(",")","......","await","subscriber.close","(",")"]
code: def __init__(
      self,
      pubsub_channel_type,
      worker_id: bytes = None,
      address: str = None,
      channel: aiogrpc.Channel = None,
  ):
      super().__init__(worker_id)
  
      if address:
          assert (
              channel is None
          ), "address and channel cannot both be specified"
          channel = gcs_utils.create_gcs_channel(address, aio=True)
      else:
          assert (
              channel is not None
          ), "One of address and channel must be specified"
      # GRPC stub to GCS pubsub.
      self._stub = gcs_service_pb2_grpc.InternalPubSubGcsServiceStub(
          channel
      )
  
      # Type of the channel.
      self._channel = pubsub_channel_type
      # A queue of received PubMessage.
      self._queue = deque()
      # Indicates whether the subscriber has closed.
      self._close = asyncio.Event()
  
imports: import asyncio
  from collections import deque
  import logging
  import random
  from typing import Tuple, List
  import grpc
  from ray._private.utils import get_or_create_event_loop
  import ray._private.gcs_utils
  import ray._private.logging_utils
  from ray.core.generated.gcs_pb2 import ErrorTableData
  from ray.core.generated import dependency_pb2
  from ray.core.generated import gcs_service_pb2_grpc
  from ray.core.generated import gcs_service_pb2
  from ray.core.generated import common_pb2
  from ray.core.generated import pubsub_pb2
label_desc: Use image node_id 1 to create a new _AioSubscriber object from inherited base classes: _SubscriberBase with example: obj = _AioSubscriber(pubsub_channel_type, worker_id, address, channel)