code: def test_actor_pubsub(
      disable_aiohttp_cache, ray_start_with_dashboard
  ):
      timeout = 5
      assert wait_until_server_available(
          ray_start_with_dashboard["webui_url"]
      )
      address_info = ray_start_with_dashboard
  
      sub = ray._raylet._TestOnly_GcsActorSubscriber(
          address=address_info["gcs_address"]
      )
      sub.subscribe()
  
      @ray.remote
      class DummyActor:
          def __init__(self):
              pass
  
      # Create a dummy actor.
      a = DummyActor.remote()
  
      def handle_pub_messages(msgs, timeout, expect_num):
          start_time = time.time()
          while (
              time.time() - start_time < timeout
              and len(msgs) < expect_num
          ):
              published = sub.poll(timeout=timeout)
              for _, actor_data in published:
                  if actor_data is None:
                      continue
                  msgs.append(actor_data)
  
      msgs = []
      handle_pub_messages(msgs, timeout, 3)
      # Assert we received published actor messages with state
      # DEPENDENCIES_UNREADY, PENDING_CREATION and ALIVE.
      assert len(msgs) == 3, msgs
  
      # Kill actor.
      ray.kill(a)
      handle_pub_messages(msgs, timeout, 4)
  
      # Assert we received published actor messages with state DEAD.
      assert len(msgs) == 4
  
      def actor_table_data_to_dict(message):
          return dashboard_utils.message_to_dict(
              message,
              {
                  "actorId",
                  "parentId",
                  "jobId",
                  "workerId",
                  "rayletId",
                  "callerId",
                  "taskId",
                  "parentTaskId",
                  "sourceActorId",
                  "placementGroupId",
              },
              including_default_value_fields=False,
          )
  
      non_state_keys = ("actorId", "jobId")
  
      for msg in msgs:
          actor_data_dict = actor_table_data_to_dict(msg)
          # DEPENDENCIES_UNREADY is 0, which would not be kept in dict. We
          # need check its original value.
          if msg.state == 0:
              assert len(actor_data_dict) > 5
              for k in non_state_keys:
                  assert k in actor_data_dict
          # For status that is not DEPENDENCIES_UNREADY, only states fields will
          # be published.
          elif actor_data_dict["state"] in ("ALIVE", "DEAD"):
              assert actor_data_dict.keys() >= {
                  "state",
                  "address",
                  "timestamp",
                  "pid",
                  "rayNamespace",
              }
          elif actor_data_dict["state"] == "PENDING_CREATION":
              assert actor_data_dict.keys() == {
                  "state",
                  "address",
                  "actorId",
                  "jobId",
                  "ownerAddress",
                  "className",
                  "serializedRuntimeEnv",
                  "rayNamespace",
                  "functionDescriptor",
                  "actorCreationDummyObjectId",
              }
          else:
              raise Exception(
                  "Unknown state: {}".format(actor_data_dict["state"])
              )
  
file_path: ray/python/ray/dashboard/modules/actor/tests/test_actor.py
imports: import logging
  import sys
  import time
  import traceback
  import pytest
  import requests
  import ray
  import ray.dashboard.utils
  from ray._private.test_utils import format_web_url, wait_until_server_available
  from ray.dashboard.modules.actor import actor_consts
  from ray.dashboard.tests.conftest import *
label_desc: Use image node_id 2 for calling a global function with example usage: test_actor_pubsub(disable_aiohttp_cache, ray_start_with_dashboard) and returns: dashboard_utils